diff --git a/docs/integrations/json-output.md b/docs/integrations/json-output.md index 5e3cff0..d446f3f 100644 --- a/docs/integrations/json-output.md +++ b/docs/integrations/json-output.md @@ -82,6 +82,14 @@ sanitizing the lane ID: "normalizer": "noop" } ], + "validator_chains": [ + { + "stage": "extract", + "lane_id": "spells", + "module_key": "dnd/spells", + "validators": [] + } + ], "validation_status": "approved", "started_at": "2026-01-01T00:00:00Z", "completed_at": "2026-01-01T00:00:01Z" @@ -103,6 +111,11 @@ references. `validation_status` is `approved` when no raw outputs were rejected and `rejected` when one or more raw outputs were rejected. +`validator_chains` records the resolved validator chain for each validation +point. Entries include stage, lane ID when applicable, module key, and validators +with key and execution class. Empty chains are recorded with an empty +`validators` array. + `normalized_outputs` summarizes each normalized lane output without embedding payload bytes. Entries include lane ID, normalizer module key, source ID, media type, and response schema provenance where available. diff --git a/docs/roadmap/validation.md b/docs/roadmap/validation.md index f5970bc..8c27355 100644 --- a/docs/roadmap/validation.md +++ b/docs/roadmap/validation.md @@ -3,17 +3,15 @@ This roadmap defines the target state for making validation a first-class, composable pipeline concern. -Current pipeline behavior is raw-output based. The runner can execute -`contracts.RawValidator` chains from `pipeline.RawValidationRegistry` for -`chunk`, `extract`, `merge`, and `normalize` outputs. Empty chains approve by +Current pipeline behavior is raw-output based. The runner validates `chunk`, +`extract`, `merge`, and `normalize` outputs through `contracts.Validator` +chains resolved from `pipeline.ValidatorChainRegistry`. Empty chains approve by default, validator rejection records a rejected raw output, and rejected output -does not pass to the next stage. Production currently registers no raw -validators, and non-empty pipeline-configured validator lists are rejected so -they cannot appear in manifests without executing. +does not pass to the next stage. Production currently registers no validators, +and non-empty pipeline-configured validator lists are rejected until +stage-scoped override syntax is implemented. -Legacy candidate validator contracts and D&D spell validators still exist under -`internal/modules/extract/dnd/spells`, but they are not part of the current -runner path. The desired end state is that validator implementations, validator +The desired end state is that validator implementations, validator registration, and default module-to-validator mappings are explicit, reviewable, and independent of concrete module packages. @@ -21,9 +19,8 @@ and independent of concrete module packages. - Move artifact and module-output validation behavior out of `internal/modules` and into `internal/validators`. -- Retire or replace the legacy candidate-oriented `contracts.Validator`, - `ValidationRequest`, and `ValidationResult` path after equivalent raw-output - validators exist. +- Keep the raw module-output `contracts.Validator`, `ValidationRequest`, and + `ValidationResult` path as the single framework validator contract. - Keep each validator in its own package. - Mirror the stage and domain shape of `internal/modules` where a validator is module-specific. @@ -261,11 +258,10 @@ The validator framework should support validation of outputs from `chunk`, enough for stage-specific validators to inspect the output they care about while ignoring irrelevant fields. -The current `contracts.RawValidationRequest` is the right starting point. It -already carries stage, lane, module, source, source and chunk provenance, -response schema metadata, raw payload, and run metadata. The final contract -should evolve from that raw-output shape rather than from the legacy -artifact-candidate `ValidationRequest`. +The current `contracts.ValidationRequest` carries stage, lane, module, source, +source and chunk provenance, response schema metadata, raw payload, and run +metadata. The final contract should continue evolving from that raw-output +shape. Additional fields needed for the full validator system include: diff --git a/internal/cli/catalog.go b/internal/cli/catalog.go index 5228d5a..72a53a2 100644 --- a/internal/cli/catalog.go +++ b/internal/cli/catalog.go @@ -20,13 +20,14 @@ import ( func productionRegistries() (pipeline.Registries, error) { registries := pipeline.Registries{ - Inputs: pipeline.NewInputAdapterRegistry(), - Chunkers: pipeline.NewChunkerRegistry(), - Extractors: pipeline.NewExtractorRegistry(), - Mergers: pipeline.NewMergerRegistry(), - Normalizers: pipeline.NewNormalizerRegistry(), - Validators: pipeline.NewValidatorRegistry(), - Outputs: pipeline.NewOutputEncoderRegistry(), + Inputs: pipeline.NewInputAdapterRegistry(), + Chunkers: pipeline.NewChunkerRegistry(), + Extractors: pipeline.NewExtractorRegistry(), + Mergers: pipeline.NewMergerRegistry(), + Normalizers: pipeline.NewNormalizerRegistry(), + Validators: pipeline.NewValidatorRegistry(), + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: pipeline.NewOutputEncoderRegistry(), } if err := seriatim.Register(registries.Inputs); err != nil { return pipeline.Registries{}, fmt.Errorf("register seriatim input: %w", err) @@ -93,25 +94,27 @@ func effectiveRegistries(opts Options) (pipeline.Registries, error) { func catalogFromRegistries(registries pipeline.Registries) pipeline.ModuleCatalog { return pipeline.ModuleCatalog{ - Inputs: registries.Inputs, - Chunkers: registries.Chunkers, - Extractors: registries.Extractors, - Mergers: registries.Mergers, - Normalizers: registries.Normalizers, - Validators: registries.Validators, - Outputs: registries.Outputs, + Inputs: registries.Inputs, + Chunkers: registries.Chunkers, + Extractors: registries.Extractors, + Mergers: registries.Mergers, + Normalizers: registries.Normalizers, + Validators: registries.Validators, + ValidatorChains: registries.ValidatorChains, + Outputs: registries.Outputs, } } func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries { return pipeline.Registries{ - Inputs: catalog.Inputs, - Chunkers: catalog.Chunkers, - Extractors: catalog.Extractors, - Mergers: catalog.Mergers, - Normalizers: catalog.Normalizers, - Validators: catalog.Validators, - Outputs: catalog.Outputs, + Inputs: catalog.Inputs, + Chunkers: catalog.Chunkers, + Extractors: catalog.Extractors, + Mergers: catalog.Mergers, + Normalizers: catalog.Normalizers, + Validators: catalog.Validators, + ValidatorChains: catalog.ValidatorChains, + Outputs: catalog.Outputs, } } @@ -122,6 +125,7 @@ func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool { catalog.Mergers == nil && catalog.Normalizers == nil && catalog.Validators == nil && + catalog.ValidatorChains == nil && catalog.Outputs == nil } @@ -132,6 +136,7 @@ func isEmptyRegistries(registries pipeline.Registries) bool { registries.Mergers == nil && registries.Normalizers == nil && registries.Validators == nil && + registries.ValidatorChains == nil && registries.Outputs == nil } diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index d63bf58..297f15b 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2931,13 +2931,14 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module mustRegisterOutput(t, outputs, specs["json"]) return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - Validators: validators, - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + Validators: validators, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } } diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index 0dd3f1b..d5f0597 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -38,10 +38,21 @@ type ArtifactLaneManifest struct { Extractor string `json:"extractor"` Merger string `json:"merger"` Normalizer string `json:"normalizer"` - Validators []string `json:"validators,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` } +type ValidatorChainManifest struct { + Stage string `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key"` + Validators []ValidatorManifest `json:"validators"` +} + +type ValidatorManifest struct { + Key string `json:"key"` + ExecutionClass string `json:"execution_class"` +} + type LLMProfileManifest struct { ID string `json:"id"` Provider string `json:"provider,omitempty"` @@ -100,6 +111,7 @@ type RunManifest struct { OutputEncoder string `json:"output_encoder,omitempty"` ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"` ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` + ValidatorChains []ValidatorChainManifest `json:"validator_chains,omitempty"` References []ReferenceProvenance `json:"references,omitempty"` NormalizedOutputs []NormalizedOutputManifest `json:"normalized_outputs,omitempty"` RejectedOutputs []RejectedOutputManifest `json:"rejected_outputs,omitempty"` diff --git a/internal/core/artifacts/artifacts_test.go b/internal/core/artifacts/artifacts_test.go index 71dc875..10eb90b 100644 --- a/internal/core/artifacts/artifacts_test.go +++ b/internal/core/artifacts/artifacts_test.go @@ -136,12 +136,21 @@ func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) { Extractor: "event-extractor", Merger: "appendorder", Normalizer: "noop", - Validators: []string{"grounded"}, Metadata: map[string]any{ "extractor": map[string]any{"prompt_id": "test.prompt"}, }, }, }, + ValidatorChains: []ValidatorChainManifest{ + { + Stage: "extract", + LaneID: "events", + ModuleKey: "event-extractor", + Validators: []ValidatorManifest{ + {Key: "grounded", ExecutionClass: "deterministic"}, + }, + }, + }, } gotJSON, err := json.Marshal(manifest) @@ -154,7 +163,7 @@ func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) { t.Fatalf("json.Unmarshal() error = %v", err) } - assertHasKeys(t, got, "pipeline_id", "pipeline_digest", "artifact_lanes", "llm_profiles") + assertHasKeys(t, got, "pipeline_id", "pipeline_digest", "artifact_lanes", "validator_chains", "llm_profiles") profiles, ok := got["llm_profiles"].([]any) if !ok { @@ -180,7 +189,20 @@ func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) { if !ok { t.Fatalf("artifact_lanes[0] = %#v, want object", lanes[0]) } - assertHasKeys(t, lane, "id", "extractor", "merger", "normalizer", "validators", "metadata") + assertHasKeys(t, lane, "id", "extractor", "merger", "normalizer", "metadata") + + chains, ok := got["validator_chains"].([]any) + if !ok { + t.Fatalf("validator_chains = %#v, want array", got["validator_chains"]) + } + if len(chains) != 1 { + t.Fatalf("len(validator_chains) = %d, want 1", len(chains)) + } + chain, ok := chains[0].(map[string]any) + if !ok { + t.Fatalf("validator_chains[0] = %#v, want object", chains[0]) + } + assertHasKeys(t, chain, "stage", "lane_id", "module_key", "validators") } func TestRunManifestIncludesReferenceProvenance(t *testing.T) { diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index cd300ce..f350455 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -429,13 +429,14 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module mustRegisterOutput(t, outputs, specs["json"]) return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - Validators: validators, - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + Validators: validators, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } } @@ -476,7 +477,8 @@ func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, func mustRegisterValidator(t *testing.T, registry *pipeline.ValidatorRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterWithSpec(spec, func() (contracts.Validator, error) { return nil, nil }); err != nil { + validatorSpec := pipeline.ValidatorSpec{Key: spec.Key, ExecutionClass: contracts.ExecutionClassDeterministic} + if err := registry.RegisterWithSpec(validatorSpec, func() (contracts.Validator, error) { return nil, nil }); err != nil { t.Fatalf("register validator: %v", err) } } diff --git a/internal/framework/pipeline/default_modules_test.go b/internal/framework/pipeline/default_modules_test.go index 0bc6aec..c909c22 100644 --- a/internal/framework/pipeline/default_modules_test.go +++ b/internal/framework/pipeline/default_modules_test.go @@ -92,12 +92,13 @@ func defaultModuleCatalog(t *testing.T) pipeline.ModuleCatalog { } return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } } diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 0e413b5..f7f2b06 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -83,6 +83,18 @@ type ResolvedArtifactLane struct { NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"` } +type ResolvedValidatorChain struct { + Stage ModuleStage `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key"` + Validators []ResolvedValidator `json:"validators"` +} + +type ResolvedValidator struct { + Binding ModuleBinding `json:"binding"` + ExecutionClass contracts.ExecutionClass `json:"execution_class"` +} + type ResolvedPipeline struct { ID string Digest string @@ -90,17 +102,19 @@ type ResolvedPipeline struct { Chunk ModuleBinding ChunkReferences ResolvedReferenceTarget `json:"chunk_references"` ArtifactLanes []ResolvedArtifactLane + ValidatorChains []ResolvedValidatorChain `json:"validator_chains"` Output ModuleBinding } type ModuleCatalog struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - Outputs *OutputEncoderRegistry + Inputs *InputAdapterRegistry + Chunkers *ChunkerRegistry + Extractors *ExtractorRegistry + Mergers *MergerRegistry + Normalizers *NormalizerRegistry + Validators *ValidatorRegistry + ValidatorChains *ValidatorChainRegistry + Outputs *OutputEncoderRegistry } func Binding(module string) ModuleBinding { @@ -172,15 +186,21 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences), Output: resolveBinding(profile.Output, DefaultOutputModule), } + chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, catalog) + if err != nil { + return ResolvedPipeline{}, err + } + resolved.ValidatorChains = append(resolved.ValidatorChains, chunkValidatorChain) outputCapabilities := capabilities.clone() for _, laneID := range selectedLaneIDs { laneProfile := lanesByID[laneID] - lane, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, profile.References, options, capabilities, catalog) + lane, validatorChains, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, profile.References, options, capabilities, catalog) if err != nil { return ResolvedPipeline{}, err } resolved.ArtifactLanes = append(resolved.ArtifactLanes, lane) + resolved.ValidatorChains = append(resolved.ValidatorChains, validatorChains...) outputCapabilities.addSet(laneCapabilities) } @@ -208,7 +228,7 @@ func resolveArtifactLane( options ResolveOptions, inherited capabilitySet, catalog ModuleCatalog, -) (ResolvedArtifactLane, capabilitySet, error) { +) (ResolvedArtifactLane, []ResolvedValidatorChain, capabilitySet, error) { lane := ResolvedArtifactLane{ ID: laneID, Extract: resolveBinding(profile.Extract, ""), @@ -217,17 +237,17 @@ func resolveArtifactLane( Validators: resolveBindings(profile.Validators, ""), } if lane.Extract.Module == "" { - return ResolvedArtifactLane{}, nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", pipelineID, laneID) + return ResolvedArtifactLane{}, nil, nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", pipelineID, laneID) } capabilities := inherited.clone() extractSpec, err := extractorSpec(catalog, lane.Extract.Module) if err != nil { - return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageExtract, lane.Extract.Module, err) + return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageExtract, lane.Extract.Module, err) } if missing, ok := capabilities.missing(extractSpec.Requires); ok { - return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) + return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) } extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References) references, err := resolveReferenceTargetBindings(referenceResolutionTarget{ @@ -241,17 +261,17 @@ func resolveArtifactLane( Options: options, }) if err != nil { - return ResolvedArtifactLane{}, nil, err + return ResolvedArtifactLane{}, nil, nil, err } lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references) capabilities.add(extractSpec.Provides...) mergeSpec, err := mergerSpec(catalog, lane.Merge.Module) if err != nil { - return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err) + return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err) } if missing, ok := capabilities.missing(mergeSpec.Requires); ok { - return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageMerge, lane.Merge.Module, missing) + return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageMerge, lane.Merge.Module, missing) } mergeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ PipelineID: pipelineID, @@ -264,17 +284,17 @@ func resolveArtifactLane( Options: options, }) if err != nil { - return ResolvedArtifactLane{}, nil, err + return ResolvedArtifactLane{}, nil, nil, err } lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences) capabilities.add(mergeSpec.Provides...) normalizeSpec, err := normalizerSpec(catalog, lane.Normalize.Module) if err != nil { - return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err) + return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err) } if missing, ok := capabilities.missing(normalizeSpec.Requires); ok { - return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing) + return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing) } normalizeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ PipelineID: pipelineID, @@ -287,22 +307,105 @@ func resolveArtifactLane( Options: options, }) if err != nil { - return ResolvedArtifactLane{}, nil, err + return ResolvedArtifactLane{}, nil, nil, err } lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences) capabilities.add(normalizeSpec.Provides...) if len(lane.Validators) > 0 { - return ResolvedArtifactLane{}, nil, configuredValidatorsError(pipelineID, laneID) + return ResolvedArtifactLane{}, nil, nil, configuredValidatorsError(pipelineID, laneID) } - return lane, capabilities, nil + extractValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageExtract, lane.Extract.Module, catalog) + if err != nil { + return ResolvedArtifactLane{}, nil, nil, err + } + mergeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageMerge, lane.Merge.Module, catalog) + if err != nil { + return ResolvedArtifactLane{}, nil, nil, err + } + normalizeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageNormalize, lane.Normalize.Module, catalog) + if err != nil { + return ResolvedArtifactLane{}, nil, nil, err + } + validatorChains := []ResolvedValidatorChain{extractValidatorChain, mergeValidatorChain, normalizeValidatorChain} + + return lane, validatorChains, capabilities, nil } func configuredValidatorsError(pipelineID string, laneID string) error { return fmt.Errorf("pipeline %q lane %q configured validators are not supported by the current raw validation runner", pipelineID, laneID) } +func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage, module string, catalog ModuleCatalog) (ResolvedValidatorChain, error) { + chain := ResolvedValidatorChain{ + Stage: stage, + LaneID: strings.TrimSpace(laneID), + ModuleKey: strings.TrimSpace(module), + } + if chain.ModuleKey == "" { + return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator chain %q module key must not be empty", pipelineID, stage) + } + switch stage { + case StageChunk, StageExtract, StageMerge, StageNormalize: + default: + return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator chain stage %q is not supported", pipelineID, stage) + } + + if catalog.ValidatorChains == nil { + return chain, nil + } + bindings := catalog.ValidatorChains.Validators(stage, chain.ModuleKey) + if len(bindings) == 0 { + return chain, nil + } + if catalog.Validators == nil { + return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator registry must not be nil for %s validator chain on module %q", pipelineID, stage, chain.ModuleKey) + } + chain.Validators = make([]ResolvedValidator, 0, len(bindings)) + for _, validator := range bindings { + spec, ok := catalog.Validators.Spec(validator.Module) + if !ok { + return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q references unknown validator %q", pipelineID, stage, chain.ModuleKey, validator.Module) + } + chain.Validators = append(chain.Validators, ResolvedValidator{ + Binding: cloneModuleBinding(validator), + ExecutionClass: spec.ExecutionClass, + }) + } + return chain, nil +} + +func cloneResolvedValidatorChains(chains []ResolvedValidatorChain) []ResolvedValidatorChain { + if len(chains) == 0 { + return nil + } + out := make([]ResolvedValidatorChain, len(chains)) + for i, chain := range chains { + out[i] = ResolvedValidatorChain{ + Stage: chain.Stage, + LaneID: strings.TrimSpace(chain.LaneID), + ModuleKey: strings.TrimSpace(chain.ModuleKey), + Validators: cloneResolvedValidators(chain.Validators), + } + } + return out +} + +func cloneResolvedValidators(validators []ResolvedValidator) []ResolvedValidator { + if len(validators) == 0 { + return nil + } + out := make([]ResolvedValidator, len(validators)) + for i, validator := range validators { + out[i] = ResolvedValidator{ + Binding: cloneModuleBinding(validator.Binding), + ExecutionClass: validator.ExecutionClass, + } + } + return out +} + func referenceTarget(stage ModuleStage, laneID string, module string, bindings []ReferenceBinding) ResolvedReferenceTarget { return ResolvedReferenceTarget{ Stage: stage, diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index b857739..9e63cea 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -110,6 +110,78 @@ func TestResolvePipelineAppliesDefaults(t *testing.T) { } } +func TestResolvePipelineRecordsValidatorChains(t *testing.T) { + catalog := newProfileCatalog(t) + if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ + Stage: StageExtract, + Module: "event-extractor", + Validators: []ModuleBinding{Binding("grounded")}, + }); err != nil { + t.Fatalf("register validator chain: %v", err) + } + + resolved, err := ResolvePipeline(PipelineProfile{ + ID: "validated", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{ + "events": {Extract: Binding("event-extractor")}, + }, + }, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + if len(resolved.ValidatorChains) != 4 { + t.Fatalf("len(ValidatorChains) = %d, want chunk plus lane extract/merge/normalize", len(resolved.ValidatorChains)) + } + extractChain := findResolvedValidatorChain(resolved.ValidatorChains, StageExtract, "events", "event-extractor") + if extractChain == nil { + t.Fatal("extract validator chain not found") + } + if len(extractChain.Validators) != 1 { + t.Fatalf("extract validators = %#v, want one validator", extractChain.Validators) + } + if extractChain.Validators[0].Binding.Module != "grounded" { + t.Fatalf("extract validator key = %q, want grounded", extractChain.Validators[0].Binding.Module) + } + if extractChain.Validators[0].ExecutionClass != contracts.ExecutionClassDeterministic { + t.Fatalf("extract validator execution class = %q, want deterministic", extractChain.Validators[0].ExecutionClass) + } + + chunkChain := findResolvedValidatorChain(resolved.ValidatorChains, StageChunk, "", DefaultChunkModule) + if chunkChain == nil { + t.Fatal("chunk validator chain not found") + } + if len(chunkChain.Validators) != 0 { + t.Fatalf("chunk validators = %#v, want explicit empty chain", chunkChain.Validators) + } +} + +func TestResolvePipelineRejectsUnknownDefaultValidator(t *testing.T) { + catalog := newProfileCatalog(t) + if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ + Stage: StageNormalize, + Module: DefaultNormalizeModule, + Validators: []ModuleBinding{Binding("missing-validator")}, + }); err != nil { + t.Fatalf("register validator chain: %v", err) + } + + _, err := ResolvePipeline(PipelineProfile{ + ID: "invalid-chain", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{ + "events": {Extract: Binding("event-extractor")}, + }, + }, ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want unknown validator error") + } + if !strings.Contains(err.Error(), "missing-validator") { + t.Fatalf("ResolvePipeline() error = %q, want missing validator context", err.Error()) + } +} + func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) { profile := multiLaneProfile() resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{" summaries ", "events", "summaries"}}, newProfileCatalog(t)) @@ -955,6 +1027,15 @@ func assertBindingSource(t *testing.T, bindings []ReferenceBinding, slotName str t.Fatalf("binding %q not found in %#v", slotName, bindings) } +func findResolvedValidatorChain(chains []ResolvedValidatorChain, stage ModuleStage, laneID string, module string) *ResolvedValidatorChain { + for i := range chains { + if chains[i].Stage == stage && chains[i].LaneID == laneID && chains[i].ModuleKey == module { + return &chains[i] + } + } + return nil +} + func newProfileCatalog(t *testing.T) ModuleCatalog { t.Helper() @@ -994,13 +1075,14 @@ func newProfileCatalogWithOverrides(t *testing.T, overrides ...ModuleSpec) Modul func emptyProfileCatalog() ModuleCatalog { return ModuleCatalog{ - Inputs: NewInputAdapterRegistry(), - Chunkers: NewChunkerRegistry(), - Extractors: NewExtractorRegistry(), - Mergers: NewMergerRegistry(), - Normalizers: NewNormalizerRegistry(), - Validators: NewValidatorRegistry(), - Outputs: NewOutputEncoderRegistry(), + Inputs: NewInputAdapterRegistry(), + Chunkers: NewChunkerRegistry(), + Extractors: NewExtractorRegistry(), + Mergers: NewMergerRegistry(), + Normalizers: NewNormalizerRegistry(), + Validators: NewValidatorRegistry(), + ValidatorChains: NewValidatorChainRegistry(), + Outputs: NewOutputEncoderRegistry(), } } @@ -1043,7 +1125,8 @@ func registerProfileSpecs(t *testing.T, catalog ModuleCatalog, specs ...ModuleSp t.Fatalf("register normalizer spec %#v: %v", spec, err) } case StageValidate: - if err := catalog.Validators.RegisterWithSpec(spec, profileValidatorConstructor(spec.Key)); err != nil { + validatorSpec := ValidatorSpec{Key: spec.Key, ExecutionClass: contracts.ExecutionClassDeterministic} + if err := catalog.Validators.RegisterWithSpec(validatorSpec, profileValidatorConstructor(spec.Key)); err != nil { t.Fatalf("register validator spec %#v: %v", spec, err) } case StageOutput: diff --git a/internal/framework/pipeline/raw_validation_registry.go b/internal/framework/pipeline/raw_validation_registry.go deleted file mode 100644 index efb0c6f..0000000 --- a/internal/framework/pipeline/raw_validation_registry.go +++ /dev/null @@ -1,78 +0,0 @@ -package pipeline - -import ( - "fmt" - "strings" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" -) - -type rawValidationKey struct { - stage ModuleStage - module string -} - -type RawValidationRegistry struct { - chains map[rawValidationKey][]contracts.Validator -} - -func NewRawValidationRegistry() *RawValidationRegistry { - return &RawValidationRegistry{ - chains: make(map[rawValidationKey][]contracts.Validator), - } -} - -func (r *RawValidationRegistry) Register(stage ModuleStage, module string, validators ...contracts.Validator) error { - if r == nil { - return fmt.Errorf("raw validation registry must not be nil") - } - normalizedModule := strings.TrimSpace(module) - if normalizedModule == "" { - return fmt.Errorf("raw validation module key must not be empty") - } - switch stage { - case StageChunk, StageExtract, StageMerge, StageNormalize: - default: - return fmt.Errorf("raw validation stage %q is not supported", stage) - } - if len(validators) == 0 { - return fmt.Errorf("raw validation chain for %q %q must not be empty", stage, normalizedModule) - } - - chain := make([]contracts.Validator, 0, len(validators)) - for i, validator := range validators { - if validator == nil { - return fmt.Errorf("raw validator %d for %q %q must not be nil", i, stage, normalizedModule) - } - if strings.TrimSpace(validator.Name()) == "" { - return fmt.Errorf("raw validator %d for %q %q must not have an empty name", i, stage, normalizedModule) - } - switch validator.ExecutionClass() { - case contracts.ExecutionClassDeterministic, contracts.ExecutionClassLLMBacked: - default: - return fmt.Errorf("raw validator %q for %q %q has unsupported execution class %q", validator.Name(), stage, normalizedModule, validator.ExecutionClass()) - } - chain = append(chain, validator) - } - - if r.chains == nil { - r.chains = make(map[rawValidationKey][]contracts.Validator) - } - key := rawValidationKey{stage: stage, module: normalizedModule} - if _, exists := r.chains[key]; exists { - return fmt.Errorf("raw validation chain for %q %q is already registered", stage, normalizedModule) - } - r.chains[key] = append([]contracts.Validator(nil), chain...) - return nil -} - -func (r *RawValidationRegistry) Validators(stage ModuleStage, module string) []contracts.Validator { - if r == nil { - return nil - } - chain := r.chains[rawValidationKey{stage: stage, module: strings.TrimSpace(module)}] - if len(chain) == 0 { - return nil - } - return append([]contracts.Validator(nil), chain...) -} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index ce2d1d7..cee362b 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -18,14 +18,14 @@ import ( ) type Registries struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - RawValidators *RawValidationRegistry - Outputs *OutputEncoderRegistry + Inputs *InputAdapterRegistry + Chunkers *ChunkerRegistry + Extractors *ExtractorRegistry + Mergers *MergerRegistry + Normalizers *NormalizerRegistry + Validators *ValidatorRegistry + ValidatorChains *ValidatorChainRegistry + Outputs *OutputEncoderRegistry } type Runner struct { @@ -127,7 +127,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if err != nil { return false, nil, fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) } - rejection, err := r.validateChunksRaw(ctx, doc, chunker.Key(), chunks, input.Metadata, attempt) + rejection, err := r.validateChunksRaw(ctx, doc, chunker.Key(), chunks, input.Metadata, input.Pipeline.ValidatorChains, attempt) if err != nil || rejection != nil { return false, rejection, err } @@ -240,6 +240,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source schema: extractOutput.Schema, payload: extractOutput.Payload, metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, attempt: attempt, }) if err != nil || rejection != nil { @@ -296,6 +297,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source schema: mergeOutput.Schema, payload: mergeOutput.Payload, metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, attempt: attempt, }) if err != nil || rejection != nil { @@ -346,6 +348,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source schema: normalizeOutput.Schema, payload: normalizeOutput.Payload, metadata: input.Metadata, + chains: input.Pipeline.ValidatorChains, attempt: attempt, }) if err != nil || rejection != nil { @@ -378,6 +381,7 @@ type rawValidationTarget struct { schema contracts.ResponseSchema payload contracts.RawPayload metadata map[string]any + chains []ResolvedValidatorChain attempt int } @@ -428,7 +432,7 @@ func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, return false, lastRejection, nil } -func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []contracts.SourceChunk, metadata map[string]any, attempt int) (*contracts.RejectedOutput, error) { +func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []contracts.SourceChunk, metadata map[string]any, chains []ResolvedValidatorChain, attempt int) (*contracts.RejectedOutput, error) { for _, chunk := range chunks { _, rejection, err := r.validateRaw(ctx, rawValidationTarget{ stage: StageChunk, @@ -443,6 +447,7 @@ func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocume Metadata: cloneMetadata(chunk.Metadata), }, metadata: metadata, + chains: chains, attempt: attempt, }) if err != nil { @@ -456,10 +461,16 @@ func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocume } func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([]contracts.Warning, *contracts.RejectedOutput, error) { - validators := r.registries.RawValidators.Validators(target.stage, target.moduleKey) - if len(validators) == 0 { + chain := resolvedValidatorChain(target.stage, target.laneID, target.moduleKey, target.chains) + if len(chain.Validators) == 0 { + chain = r.registryValidatorChain(target.stage, target.laneID, target.moduleKey) + } + if len(chain.Validators) == 0 { return nil, nil, nil } + if r.registries.Validators == nil { + return nil, nil, fmt.Errorf("validator registry must not be nil") + } request := contracts.ValidationRequest{ Stage: string(target.stage), @@ -475,7 +486,11 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ } var warnings []contracts.Warning - for _, validator := range validators { + for _, validatorBinding := range chain.Validators { + validator, err := r.registries.Validators.Build(validatorBinding.Binding.Module) + if err != nil { + return nil, nil, fmt.Errorf("build validator %q: %w", validatorBinding.Binding.Module, err) + } result, err := validator.Validate(ctx, request) if err != nil { return nil, nil, fmt.Errorf("validate raw %s output with validator %q: %w", target.stage, validator.Name(), err) @@ -507,6 +522,60 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ return warnings, nil, nil } +func (r *Runner) registryValidatorChain(stage ModuleStage, laneID string, moduleKey string) ResolvedValidatorChain { + chain := ResolvedValidatorChain{ + Stage: stage, + LaneID: strings.TrimSpace(laneID), + ModuleKey: strings.TrimSpace(moduleKey), + } + if r == nil || r.registries.ValidatorChains == nil { + return chain + } + bindings := r.registries.ValidatorChains.Validators(stage, moduleKey) + if len(bindings) == 0 { + return chain + } + chain.Validators = make([]ResolvedValidator, 0, len(bindings)) + for _, binding := range bindings { + executionClass := contracts.ExecutionClass("") + if r.registries.Validators != nil { + if spec, ok := r.registries.Validators.Spec(binding.Module); ok { + executionClass = spec.ExecutionClass + } + } + chain.Validators = append(chain.Validators, ResolvedValidator{ + Binding: cloneModuleBinding(binding), + ExecutionClass: executionClass, + }) + } + return chain +} + +func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain { + for _, chain := range chains { + if chain.Stage != stage { + continue + } + if chain.ModuleKey != moduleKey { + continue + } + if strings.TrimSpace(chain.LaneID) != strings.TrimSpace(laneID) { + continue + } + return ResolvedValidatorChain{ + Stage: chain.Stage, + LaneID: chain.LaneID, + ModuleKey: chain.ModuleKey, + Validators: cloneResolvedValidators(chain.Validators), + } + } + return ResolvedValidatorChain{ + Stage: stage, + LaneID: strings.TrimSpace(laneID), + ModuleKey: strings.TrimSpace(moduleKey), + } +} + func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error { if r.registries.Inputs == nil { return fmt.Errorf("input registry must not be nil") @@ -580,16 +649,17 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { pipeline := input.Pipeline manifest := artifacts.RunManifest{ - PipelineID: pipeline.ID, - PipelineDigest: pipeline.Digest, - InputModule: pipeline.Input.Module, - Chunker: pipeline.Chunk.Module, - OutputEncoder: pipeline.Output.Module, - ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)), - RunID: runID, - StartedAt: timePtr(startedAt), - References: ReferenceProvenance(pipeline), - LLMProfiles: cloneLLMProfiles(input.LLMProfiles), + PipelineID: pipeline.ID, + PipelineDigest: pipeline.Digest, + InputModule: pipeline.Input.Module, + Chunker: pipeline.Chunk.Module, + OutputEncoder: pipeline.Output.Module, + ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)), + ValidatorChains: validatorChainManifests(pipeline.ValidatorChains), + RunID: runID, + StartedAt: timePtr(startedAt), + References: ReferenceProvenance(pipeline), + LLMProfiles: cloneLLMProfiles(input.LLMProfiles), } // The runner does not currently maintain a cache or idempotency key. Reference // digests are recorded in manifest provenance and intentionally kept separate @@ -607,6 +677,29 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { return manifest } +func validatorChainManifests(chains []ResolvedValidatorChain) []artifacts.ValidatorChainManifest { + if len(chains) == 0 { + return nil + } + manifests := make([]artifacts.ValidatorChainManifest, 0, len(chains)) + for _, chain := range chains { + manifest := artifacts.ValidatorChainManifest{ + Stage: string(chain.Stage), + LaneID: chain.LaneID, + ModuleKey: chain.ModuleKey, + Validators: make([]artifacts.ValidatorManifest, 0, len(chain.Validators)), + } + for _, validator := range chain.Validators { + manifest.Validators = append(manifest.Validators, artifacts.ValidatorManifest{ + Key: validator.Binding.Module, + ExecutionClass: string(validator.ExecutionClass), + }) + } + manifests = append(manifests, manifest) + } + return manifests +} + func failOutput(output RunOutput) RunOutput { if output.Manifest.PipelineID != "" { populateRawOutputManifest(&output) diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 6f6d580..91080f2 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -887,8 +887,9 @@ func TestRunPassesChunkContentAndMediaTypeToExtractors(t *testing.T) { func TestRunOmitsRejectedExtractOutputsFromMerge(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-extract", approved: []bool{false, true}, reason: "bad_extract", message: "extract rejected"} - modules.rawValidators = rawValidationRegistry(t, StageExtract, "extract-alpha", validator) + validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false, true}, reason: "bad_extract", message: "extract rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageExtract, "extract-alpha", validator) output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) if err != nil { @@ -909,8 +910,9 @@ func TestRunOmitsRejectedExtractOutputsFromMerge(t *testing.T) { func TestRunOmitsLaneWithNoAcceptedExtractOutputs(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-extract", approved: []bool{false}, reason: "bad_extract", message: "extract rejected"} - modules.rawValidators = rawValidationRegistry(t, StageExtract, "extract-alpha", validator) + validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false}, reason: "bad_extract", message: "extract rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageExtract, "extract-alpha", validator) output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) if err != nil { @@ -933,8 +935,9 @@ func TestRunOmitsLaneWithNoAcceptedExtractOutputs(t *testing.T) { func TestRunRejectedMergePreventsNormalizeForLane(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-merge", approved: []bool{false}, reason: "bad_merge", message: "merge rejected"} - modules.rawValidators = rawValidationRegistry(t, StageMerge, "merge", validator) + validator := &runnerChainValidator{name: "chain-merge", approved: []bool{false}, reason: "bad_merge", message: "merge rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageMerge, "merge", validator) output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) if err != nil { @@ -954,8 +957,9 @@ func TestRunRejectedMergePreventsNormalizeForLane(t *testing.T) { func TestRunRejectedNormalizePreventsOutputForLane(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-normalize", approved: []bool{false}, reason: "bad_normalize", message: "normalize rejected"} - modules.rawValidators = rawValidationRegistry(t, StageNormalize, "normalize", validator) + validator := &runnerChainValidator{name: "chain-normalize", approved: []bool{false}, reason: "bad_normalize", message: "normalize rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageNormalize, "normalize", validator) output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) if err != nil { @@ -999,8 +1003,9 @@ func TestRunRetriesSameModuleInputAfterFrameworkError(t *testing.T) { func TestRunRetriesSameModuleInputAfterValidatorRejection(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-extract", approved: []bool{false, true, true}, reason: "bad_extract", message: "extract rejected"} - modules.rawValidators = rawValidationRegistry(t, StageExtract, "extract-alpha", validator) + validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false, true, true}, reason: "bad_extract", message: "extract rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageExtract, "extract-alpha", validator) pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].Extract.Retries = 1 @@ -1023,8 +1028,9 @@ func TestRunRetriesSameModuleInputAfterValidatorRejection(t *testing.T) { func TestRunStopsRetryAfterConfiguredAttemptsAndRecordsAttemptCount(t *testing.T) { modules := defaultRunnerModules() - validator := &runnerRawValidator{name: "raw-extract", approved: []bool{false}, reason: "bad_extract", message: "extract rejected"} - modules.rawValidators = rawValidationRegistry(t, StageExtract, "extract-alpha", validator) + validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false}, reason: "bad_extract", message: "extract rejected"} + modules.validators[validator.name] = validator + modules.validatorChains = validatorChainRegistry(t, StageExtract, "extract-alpha", validator) pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].Extract.Retries = 1 @@ -1312,8 +1318,11 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) { if lane.ID != "alpha" || lane.Extractor != "extract-alpha" || lane.Merger != "merge" || lane.Normalizer != "normalize" { t.Fatalf("ArtifactLanes[0] = %#v, want lane details", lane) } - if len(lane.Validators) != 0 { - t.Fatalf("lane validators = %#v, want none", lane.Validators) + if len(manifest.ValidatorChains) != 4 { + t.Fatalf("ValidatorChains = %#v, want four validation points", manifest.ValidatorChains) + } + if manifest.ValidatorChains[0].Stage != string(StageChunk) || manifest.ValidatorChains[0].ModuleKey != "chunk" || len(manifest.ValidatorChains[0].Validators) != 0 { + t.Fatalf("chunk validator chain = %#v, want explicit empty chunk chain", manifest.ValidatorChains[0]) } } @@ -1454,6 +1463,12 @@ func resolvedPipeline() ResolvedPipeline { NormalizeReferences: referenceTarget(StageNormalize, "alpha", "normalize", nil), }, }, + ValidatorChains: []ResolvedValidatorChain{ + {Stage: StageChunk, ModuleKey: "chunk"}, + {Stage: StageExtract, LaneID: "alpha", ModuleKey: "extract-alpha"}, + {Stage: StageMerge, LaneID: "alpha", ModuleKey: "merge"}, + {Stage: StageNormalize, LaneID: "alpha", ModuleKey: "normalize"}, + }, Output: Binding("output"), } } @@ -1493,8 +1508,8 @@ type runnerModules struct { extractors map[string]*runnerExtractor mergers map[string]*runnerMerger normalizers map[string]*runnerNormalizer - validators map[string]*runnerValidator - rawValidators *RawValidationRegistry + validators map[string]contracts.Validator + validatorChains *ValidatorChainRegistry output *runnerOutputEncoder inputBuildErr error chunkerBuildErr error @@ -1513,9 +1528,9 @@ func defaultRunnerModules() *runnerModules { normalizers: map[string]*runnerNormalizer{ "normalize": {key: "normalize"}, }, - validators: map[string]*runnerValidator{ - "configured": {name: "configured"}, - "second-validator": {name: "second-validator"}, + validators: map[string]contracts.Validator{ + "configured": &runnerValidator{name: "configured"}, + "second-validator": &runnerValidator{name: "second-validator"}, }, output: &runnerOutputEncoder{ key: "output", @@ -1533,14 +1548,14 @@ func newRunnerRegistries(t *testing.T, modules *runnerModules) Registries { } registries := Registries{ - Inputs: NewInputAdapterRegistry(), - Chunkers: NewChunkerRegistry(), - Extractors: NewExtractorRegistry(), - Mergers: NewMergerRegistry(), - Normalizers: NewNormalizerRegistry(), - Validators: NewValidatorRegistry(), - RawValidators: modules.rawValidators, - Outputs: NewOutputEncoderRegistry(), + Inputs: NewInputAdapterRegistry(), + Chunkers: NewChunkerRegistry(), + Extractors: NewExtractorRegistry(), + Mergers: NewMergerRegistry(), + Normalizers: NewNormalizerRegistry(), + Validators: NewValidatorRegistry(), + ValidatorChains: modules.validatorChains, + Outputs: NewOutputEncoderRegistry(), } if err := registries.Inputs.Register("input", func() (contracts.InputAdapter, error) { if modules.inputBuildErr != nil { @@ -1807,7 +1822,7 @@ type runnerValidator struct { requests []contracts.ValidationRequest } -type runnerRawValidator struct { +type runnerChainValidator struct { name string approved []bool reason string @@ -1818,15 +1833,15 @@ type runnerRawValidator struct { requests []contracts.ValidationRequest } -func (validator *runnerRawValidator) Name() string { +func (validator *runnerChainValidator) Name() string { return validator.name } -func (validator *runnerRawValidator) ExecutionClass() contracts.ExecutionClass { +func (validator *runnerChainValidator) ExecutionClass() contracts.ExecutionClass { return contracts.ExecutionClassDeterministic } -func (validator *runnerRawValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) { +func (validator *runnerChainValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) { validator.calls++ validator.requests = append(validator.requests, req) if validator.err != nil { @@ -2037,12 +2052,16 @@ func assertRunError(t *testing.T, err error, want string) { } } -func rawValidationRegistry(t *testing.T, stage ModuleStage, module string, validators ...contracts.Validator) *RawValidationRegistry { +func validatorChainRegistry(t *testing.T, stage ModuleStage, module string, validators ...contracts.Validator) *ValidatorChainRegistry { t.Helper() - registry := NewRawValidationRegistry() - if err := registry.Register(stage, module, validators...); err != nil { - t.Fatalf("register raw validators: %v", err) + registry := NewValidatorChainRegistry() + bindings := make([]ModuleBinding, 0, len(validators)) + for _, validator := range validators { + bindings = append(bindings, Binding(validator.Name())) + } + if err := registry.Register(ValidatorChainMapping{Stage: stage, Module: module, Validators: bindings}); err != nil { + t.Fatalf("register validator chain: %v", err) } return registry } diff --git a/internal/framework/pipeline/validator_chain_registry.go b/internal/framework/pipeline/validator_chain_registry.go new file mode 100644 index 0000000..0e8882e --- /dev/null +++ b/internal/framework/pipeline/validator_chain_registry.go @@ -0,0 +1,102 @@ +package pipeline + +import ( + "fmt" + "strings" +) + +type validatorChainKey struct { + stage ModuleStage + module string +} + +type ValidatorChainMapping struct { + Stage ModuleStage `json:"stage"` + Module string `json:"module"` + Validators []ModuleBinding `json:"validators,omitempty"` +} + +type ValidatorChainRegistry struct { + chains map[validatorChainKey][]ModuleBinding +} + +func NewValidatorChainRegistry() *ValidatorChainRegistry { + return &ValidatorChainRegistry{ + chains: make(map[validatorChainKey][]ModuleBinding), + } +} + +func (r *ValidatorChainRegistry) Register(mapping ValidatorChainMapping) error { + if r == nil { + return fmt.Errorf("validator chain registry must not be nil") + } + normalized, err := normalizeValidatorChainMapping(mapping) + if err != nil { + return err + } + if r.chains == nil { + r.chains = make(map[validatorChainKey][]ModuleBinding) + } + key := validatorChainKey{stage: normalized.Stage, module: normalized.Module} + if _, exists := r.chains[key]; exists { + return fmt.Errorf("validator chain for %q %q is already registered", normalized.Stage, normalized.Module) + } + r.chains[key] = cloneModuleBindings(normalized.Validators) + return nil +} + +func (r *ValidatorChainRegistry) Validators(stage ModuleStage, module string) []ModuleBinding { + if r == nil { + return nil + } + chain := r.chains[validatorChainKey{stage: stage, module: strings.TrimSpace(module)}] + return cloneModuleBindings(chain) +} + +func normalizeValidatorChainMapping(mapping ValidatorChainMapping) (ValidatorChainMapping, error) { + normalized := ValidatorChainMapping{ + Stage: mapping.Stage, + Module: strings.TrimSpace(mapping.Module), + Validators: cloneModuleBindings(mapping.Validators), + } + switch normalized.Stage { + case StageChunk, StageExtract, StageMerge, StageNormalize: + default: + return ValidatorChainMapping{}, fmt.Errorf("validator chain stage %q is not supported", normalized.Stage) + } + if normalized.Module == "" { + return ValidatorChainMapping{}, fmt.Errorf("validator chain module key must not be empty") + } + for i, validator := range normalized.Validators { + if strings.TrimSpace(validator.Module) == "" { + return ValidatorChainMapping{}, fmt.Errorf("validator chain for %q %q has empty validator key at index %d", normalized.Stage, normalized.Module, i) + } + normalized.Validators[i] = resolveBinding(validator, "") + } + return normalized, nil +} + +func cloneModuleBindings(bindings []ModuleBinding) []ModuleBinding { + if len(bindings) == 0 { + return nil + } + out := make([]ModuleBinding, len(bindings)) + for i, binding := range bindings { + out[i] = cloneModuleBinding(binding) + } + return out +} + +func cloneModuleBinding(binding ModuleBinding) ModuleBinding { + binding.Module = strings.TrimSpace(binding.Module) + binding.LLMProfile = strings.TrimSpace(binding.LLMProfile) + binding.Options = cloneOptions(binding.Options) + if len(binding.References) > 0 { + references := make(map[string]string, len(binding.References)) + for key, value := range binding.References { + references[key] = value + } + binding.References = references + } + return binding +} diff --git a/internal/framework/pipeline/validator_chain_registry_test.go b/internal/framework/pipeline/validator_chain_registry_test.go new file mode 100644 index 0000000..057c555 --- /dev/null +++ b/internal/framework/pipeline/validator_chain_registry_test.go @@ -0,0 +1,74 @@ +package pipeline + +import "testing" + +func TestValidatorChainRegistryRegistersAndLooksUpChains(t *testing.T) { + registry := NewValidatorChainRegistry() + err := registry.Register(ValidatorChainMapping{ + Stage: StageExtract, + Module: " extractor ", + Validators: []ModuleBinding{{Module: " first "}, {Module: "second"}}, + }) + if err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + + got := registry.Validators(StageExtract, " extractor ") + if len(got) != 2 || got[0].Module != "first" || got[1].Module != "second" { + t.Fatalf("Validators() = %#v, want trimmed chain", got) + } +} + +func TestValidatorChainRegistryRejectsDuplicateMappings(t *testing.T) { + registry := NewValidatorChainRegistry() + mapping := ValidatorChainMapping{Stage: StageMerge, Module: "merge", Validators: []ModuleBinding{Binding("validator")}} + if err := registry.Register(mapping); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + + if err := registry.Register(mapping); err == nil { + t.Fatal("Register() error = nil, want duplicate mapping error") + } +} + +func TestValidatorChainRegistryRejectsUnsupportedStage(t *testing.T) { + registry := NewValidatorChainRegistry() + err := registry.Register(ValidatorChainMapping{Stage: StageInput, Module: "input"}) + if err == nil { + t.Fatal("Register() error = nil, want unsupported stage error") + } +} + +func TestValidatorChainRegistryAllowsAbsentAndEmptyChains(t *testing.T) { + registry := NewValidatorChainRegistry() + if got := registry.Validators(StageNormalize, "normalize"); got != nil { + t.Fatalf("absent chain = %#v, want nil", got) + } + if err := registry.Register(ValidatorChainMapping{Stage: StageNormalize, Module: "normalize"}); err != nil { + t.Fatalf("Register(empty) error = %v, want nil", err) + } + if got := registry.Validators(StageNormalize, "normalize"); got != nil { + t.Fatalf("empty chain = %#v, want nil", got) + } +} + +func TestValidatorChainRegistryReturnsDefensiveCopies(t *testing.T) { + registry := NewValidatorChainRegistry() + err := registry.Register(ValidatorChainMapping{ + Stage: StageChunk, + Module: "chunk", + Validators: []ModuleBinding{{Module: "validator", Options: map[string]any{"level": "strict"}}}, + }) + if err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + + got := registry.Validators(StageChunk, "chunk") + got[0].Module = "changed" + got[0].Options["level"] = "changed" + + again := registry.Validators(StageChunk, "chunk") + if again[0].Module != "validator" || again[0].Options["level"] != "strict" { + t.Fatalf("Validators() after caller mutation = %#v, want original chain", again) + } +} diff --git a/internal/framework/pipeline/validator_registry.go b/internal/framework/pipeline/validator_registry.go index 17e6b15..c8e0cad 100644 --- a/internal/framework/pipeline/validator_registry.go +++ b/internal/framework/pipeline/validator_registry.go @@ -2,6 +2,7 @@ package pipeline import ( "fmt" + "sort" "strings" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" @@ -9,29 +10,34 @@ import ( type ValidatorConstructor func() (contracts.Validator, error) +type ValidatorSpec struct { + Key string `json:"key"` + ExecutionClass contracts.ExecutionClass `json:"execution_class"` +} + type ValidatorRegistry struct { constructors map[string]ValidatorConstructor - specs map[string]ModuleSpec + specs map[string]ValidatorSpec } func NewValidatorRegistry() *ValidatorRegistry { return &ValidatorRegistry{ constructors: make(map[string]ValidatorConstructor), - specs: make(map[string]ModuleSpec), + specs: make(map[string]ValidatorSpec), } } func (r *ValidatorRegistry) Register(key string, constructor ValidatorConstructor) error { - return r.RegisterWithSpec(defaultModuleSpec(key, StageValidate), constructor) + return r.RegisterWithSpec(ValidatorSpec{Key: key, ExecutionClass: contracts.ExecutionClassDeterministic}, constructor) } -func (r *ValidatorRegistry) RegisterWithSpec(spec ModuleSpec, constructor ValidatorConstructor) error { +func (r *ValidatorRegistry) RegisterWithSpec(spec ValidatorSpec, constructor ValidatorConstructor) error { if r == nil { return fmt.Errorf("validator registry must not be nil") } - normalizedSpec := normalizeModuleSpec(spec) - if err := validateModuleSpec("validator", StageValidate, normalizedSpec); err != nil { + normalizedSpec, err := normalizeValidatorSpec(spec) + if err != nil { return err } if constructor == nil { @@ -45,10 +51,10 @@ func (r *ValidatorRegistry) RegisterWithSpec(spec ModuleSpec, constructor Valida r.constructors = make(map[string]ValidatorConstructor) } if r.specs == nil { - r.specs = make(map[string]ModuleSpec) + r.specs = make(map[string]ValidatorSpec) } r.constructors[normalizedSpec.Key] = constructor - r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) + r.specs[normalizedSpec.Key] = normalizedSpec return nil } @@ -77,25 +83,45 @@ func (r *ValidatorRegistry) Build(key string) (contracts.Validator, error) { if validator.Name() != normalizedKey { return nil, fmt.Errorf("validator %q returned name %q", normalizedKey, validator.Name()) } - switch validator.ExecutionClass() { - case contracts.ExecutionClassDeterministic, contracts.ExecutionClassLLMBacked: - default: - return nil, fmt.Errorf("validator %q returned unsupported execution class %q", normalizedKey, validator.ExecutionClass()) + spec, ok := r.specs[normalizedKey] + if !ok { + return nil, fmt.Errorf("validator %q spec is not registered", normalizedKey) + } + if validator.ExecutionClass() != spec.ExecutionClass { + return nil, fmt.Errorf("validator %q returned execution class %q, want %q", normalizedKey, validator.ExecutionClass(), spec.ExecutionClass) } return validator, nil } -func (r *ValidatorRegistry) Spec(key string) (ModuleSpec, bool) { +func (r *ValidatorRegistry) Spec(key string) (ValidatorSpec, bool) { if r == nil { - return ModuleSpec{}, false + return ValidatorSpec{}, false } spec, ok := r.specs[strings.TrimSpace(key)] if !ok { - return ModuleSpec{}, false + return ValidatorSpec{}, false } - return cloneModuleSpec(spec), true + return spec, true +} + +func (r *ValidatorRegistry) RegisteredSpecs() []ValidatorSpec { + if r == nil || len(r.specs) == 0 { + return nil + } + + keys := make([]string, 0, len(r.specs)) + for key := range r.specs { + keys = append(keys, key) + } + sort.Strings(keys) + + specs := make([]ValidatorSpec, 0, len(keys)) + for _, key := range keys { + specs = append(specs, r.specs[key]) + } + return specs } func (r *ValidatorRegistry) RegisteredKeys() []string { @@ -105,3 +131,19 @@ func (r *ValidatorRegistry) RegisteredKeys() []string { return sortedRegistryKeys(r.constructors) } + +func normalizeValidatorSpec(spec ValidatorSpec) (ValidatorSpec, error) { + normalized := ValidatorSpec{ + Key: strings.TrimSpace(spec.Key), + ExecutionClass: spec.ExecutionClass, + } + if normalized.Key == "" { + return ValidatorSpec{}, fmt.Errorf("validator key must not be empty") + } + switch normalized.ExecutionClass { + case contracts.ExecutionClassDeterministic, contracts.ExecutionClassLLMBacked: + default: + return ValidatorSpec{}, fmt.Errorf("validator %q execution class %q is not supported", normalized.Key, normalized.ExecutionClass) + } + return normalized, nil +} diff --git a/internal/framework/pipeline/validator_registry_test.go b/internal/framework/pipeline/validator_registry_test.go index a20d497..7b20fe6 100644 --- a/internal/framework/pipeline/validator_registry_test.go +++ b/internal/framework/pipeline/validator_registry_test.go @@ -2,88 +2,116 @@ package pipeline import ( "context" + "reflect" + "strings" "testing" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) func TestValidatorRegistryBehavior(t *testing.T) { - runRegistryBehaviorTests(t, registryBehaviorCase[contracts.Validator]{ - name: "ValidatorRegistry", - key: "generic-validator", - stage: StageValidate, - wrongStage: StageExtract, - newRegistry: func() any { - return NewValidatorRegistry() - }, - register: func(registry any, key string, constructor func() (contracts.Validator, error)) error { - return registry.(*ValidatorRegistry).Register(key, constructor) - }, - registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.Validator, error)) error { - return registry.(*ValidatorRegistry).RegisterWithSpec(spec, constructor) - }, - build: func(registry any, key string) (contracts.Validator, error) { - return registry.(*ValidatorRegistry).Build(key) - }, - spec: func(registry any, key string) (ModuleSpec, bool) { - return registry.(*ValidatorRegistry).Spec(key) - }, - registeredKeys: func(registry any) []string { - return registry.(*ValidatorRegistry).RegisteredKeys() - }, - nilRegister: func(key string, constructor func() (contracts.Validator, error)) error { - var registry *ValidatorRegistry - return registry.Register(key, constructor) - }, - nilBuild: func(key string) (contracts.Validator, error) { - var registry *ValidatorRegistry - return registry.Build(key) - }, - nilSpec: func(key string) (ModuleSpec, bool) { - var registry *ValidatorRegistry - return registry.Spec(key) - }, - nilRegisteredKey: func() []string { - var registry *ValidatorRegistry - return registry.RegisteredKeys() - }, - constructor: func(key string) func() (contracts.Validator, error) { - return func() (contracts.Validator, error) { - return registryValidator{name: key}, nil - } - }, - moduleKey: func(module contracts.Validator) string { - return module.Name() - }, - }) + registry := NewValidatorRegistry() + if err := registry.Register(" generic-validator ", validatorConstructor("generic-validator", contracts.ExecutionClassDeterministic)); err != nil { + t.Fatalf("Register() error = %v, want nil", err) + } + + validator, err := registry.Build("generic-validator") + if err != nil { + t.Fatalf("Build() error = %v, want nil", err) + } + if validator.Name() != "generic-validator" { + t.Fatalf("validator name = %q, want generic-validator", validator.Name()) + } + + spec, ok := registry.Spec(" generic-validator ") + if !ok { + t.Fatal("Spec() ok = false, want true") + } + want := ValidatorSpec{Key: "generic-validator", ExecutionClass: contracts.ExecutionClassDeterministic} + if !reflect.DeepEqual(spec, want) { + t.Fatalf("Spec() = %#v, want %#v", spec, want) + } +} + +func TestValidatorRegistryRegistersSpecs(t *testing.T) { + registry := NewValidatorRegistry() + spec := ValidatorSpec{Key: " llm-validator ", ExecutionClass: contracts.ExecutionClassLLMBacked} + if err := registry.RegisterWithSpec(spec, validatorConstructor("llm-validator", contracts.ExecutionClassLLMBacked)); err != nil { + t.Fatalf("RegisterWithSpec() error = %v, want nil", err) + } + + got, ok := registry.Spec("llm-validator") + if !ok { + t.Fatal("Spec() ok = false, want true") + } + want := ValidatorSpec{Key: "llm-validator", ExecutionClass: contracts.ExecutionClassLLMBacked} + if !reflect.DeepEqual(got, want) { + t.Fatalf("Spec() = %#v, want %#v", got, want) + } +} + +func TestValidatorRegistryRegisteredSpecsAreSorted(t *testing.T) { + registry := NewValidatorRegistry() + for _, key := range []string{"zeta", "alpha"} { + if err := registry.Register(key, validatorConstructor(key, contracts.ExecutionClassDeterministic)); err != nil { + t.Fatalf("Register(%q) error = %v", key, err) + } + } + + specs := registry.RegisteredSpecs() + if len(specs) != 2 || specs[0].Key != "alpha" || specs[1].Key != "zeta" { + t.Fatalf("RegisteredSpecs() = %#v, want sorted specs", specs) + } } func TestValidatorRegistryRejectsUnsupportedExecutionClass(t *testing.T) { registry := NewValidatorRegistry() - if err := registry.Register("invalid-validator", func() (contracts.Validator, error) { - return invalidExecutionClassValidator{name: "invalid-validator"}, nil - }); err != nil { - t.Fatalf("Register() error = %v, want nil", err) - } - - _, err := registry.Build("invalid-validator") + err := registry.RegisterWithSpec( + ValidatorSpec{Key: "invalid-validator", ExecutionClass: contracts.ExecutionClass("unsupported")}, + validatorConstructor("invalid-validator", contracts.ExecutionClass("unsupported")), + ) if err == nil { - t.Fatal("Build() error = nil, want unsupported execution class error") + t.Fatal("RegisterWithSpec() error = nil, want unsupported execution class error") } } -type invalidExecutionClassValidator struct { - name string +func TestValidatorRegistryRejectsConstructorExecutionClassMismatch(t *testing.T) { + registry := NewValidatorRegistry() + if err := registry.RegisterWithSpec( + ValidatorSpec{Key: "validator", ExecutionClass: contracts.ExecutionClassDeterministic}, + validatorConstructor("validator", contracts.ExecutionClassLLMBacked), + ); err != nil { + t.Fatalf("RegisterWithSpec() error = %v, want nil", err) + } + + _, err := registry.Build("validator") + if err == nil { + t.Fatal("Build() error = nil, want execution class mismatch") + } + if !strings.Contains(err.Error(), "execution class") { + t.Fatalf("Build() error = %q, want execution class context", err.Error()) + } } -func (validator invalidExecutionClassValidator) Name() string { +type testValidator struct { + name string + executionClass contracts.ExecutionClass +} + +func validatorConstructor(name string, executionClass contracts.ExecutionClass) ValidatorConstructor { + return func() (contracts.Validator, error) { + return testValidator{name: name, executionClass: executionClass}, nil + } +} + +func (validator testValidator) Name() string { return validator.name } -func (validator invalidExecutionClassValidator) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClass("unsupported") +func (validator testValidator) ExecutionClass() contracts.ExecutionClass { + return validator.executionClass } -func (validator invalidExecutionClassValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) { +func (validator testValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) { return contracts.ValidationResult{Approved: true}, nil } diff --git a/internal/framework/pipeline/walking_skeleton_test.go b/internal/framework/pipeline/walking_skeleton_test.go index f0771af..ca03106 100644 --- a/internal/framework/pipeline/walking_skeleton_test.go +++ b/internal/framework/pipeline/walking_skeleton_test.go @@ -97,12 +97,13 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog { t.Helper() catalog := ModuleCatalog{ - Inputs: NewInputAdapterRegistry(), - Chunkers: NewChunkerRegistry(), - Extractors: NewExtractorRegistry(), - Mergers: NewMergerRegistry(), - Normalizers: NewNormalizerRegistry(), - Outputs: NewOutputEncoderRegistry(), + Inputs: NewInputAdapterRegistry(), + Chunkers: NewChunkerRegistry(), + Extractors: NewExtractorRegistry(), + Mergers: NewMergerRegistry(), + Normalizers: NewNormalizerRegistry(), + ValidatorChains: NewValidatorChainRegistry(), + Outputs: NewOutputEncoderRegistry(), } if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{ Key: "fake/input", diff --git a/internal/modules/extract/dnd/spells/config_test.go b/internal/modules/extract/dnd/spells/config_test.go index c12c31f..20c44c7 100644 --- a/internal/modules/extract/dnd/spells/config_test.go +++ b/internal/modules/extract/dnd/spells/config_test.go @@ -212,12 +212,13 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo } return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } } diff --git a/internal/modules/input/seriatim/config_test.go b/internal/modules/input/seriatim/config_test.go index 334a381..7dce6b2 100644 --- a/internal/modules/input/seriatim/config_test.go +++ b/internal/modules/input/seriatim/config_test.go @@ -151,12 +151,13 @@ func seriatimTestCatalog(t *testing.T, inputSpec pipeline.ModuleSpec) pipeline.M }) return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } }