Add validator chain config overrides

internal/core/config/config.go

@@ -97,6 +97,18 @@ func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding {
 		out.Options = cloneOptions(in.Options)
 	}
 	out.References = cloneStringMap(in.References)
+	out.Validators = cloneValidatorOverride(in.Validators)
+	return out
+}
+
+func cloneValidatorOverride(in pipeline.ValidatorOverride) pipeline.ValidatorOverride {
+	out := pipeline.ValidatorOverride{Set: in.Set}
+	if len(in.Validators) > 0 {
+		out.Validators = make([]pipeline.ModuleBinding, len(in.Validators))
+		for i, binding := range in.Validators {
+			out.Validators[i] = cloneModuleBinding(binding)
+		}
+	}
 	return out
 }
 

internal/core/config/effective_config_test.go

@@ -161,6 +161,12 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) {
 	profile.Output.LLMProfile = "output-profile"
 	lane := profile.Artifacts["events"]
 	lane.Merge.LLMProfile = "merge-profile"
+	lane.Extract.Validators = pipeline.ValidatorOverride{
+		Set: true,
+		Validators: []pipeline.ModuleBinding{
+			{Module: "fake/llm-validator", LLMProfile: "validator-profile"},
+		},
+	}
 	profile.Artifacts["events"] = lane
 	cfg.Pipelines["example"] = profile
 
@@ -195,9 +201,22 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) {
 	if eventLane.Merge.LLMProfile != "runtime" {
 		t.Fatalf("merge profile = %q, want runtime", eventLane.Merge.LLMProfile)
 	}
-	if len(eventLane.Validators) != 0 {
-		t.Fatalf("validator profiles = %#v, want none", eventLane.Validators)
+	validatorChain := findEffectiveValidatorChain(effective.ResolvedPipeline.ValidatorChains, pipeline.StageExtract, "events", "fake/extract")
+	if validatorChain == nil || len(validatorChain.Validators) != 1 {
+		t.Fatalf("validator chain = %#v, want one extract validator", effective.ResolvedPipeline.ValidatorChains)
 	}
+	if validatorChain.Validators[0].Binding.LLMProfile != "validator-profile" {
+		t.Fatalf("validator profile = %q, want original validator-profile", validatorChain.Validators[0].Binding.LLMProfile)
+	}
+}
+
+func findEffectiveValidatorChain(chains []pipeline.ResolvedValidatorChain, stage pipeline.ModuleStage, laneID string, module string) *pipeline.ResolvedValidatorChain {
+	for i := range chains {
+		if chains[i].Stage == stage && chains[i].LaneID == laneID && chains[i].ModuleKey == module {
+			return &chains[i]
+		}
+	}
+	return nil
 }
 
 func llmCapableBindings(resolved pipeline.ResolvedPipeline) []pipeline.ModuleBinding {

internal/core/config/file_config.go

@@ -56,6 +56,7 @@ type fileModuleBinding struct {
 	Retries    int
 	Options    map[string]any
 	References map[string]string
+	Validators pipeline.ValidatorOverride
 }
 
 func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
@@ -102,6 +103,16 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
 					return err
 				}
 				b.References = references
+			case "validators":
+				b.Validators.Set = true
+				var validators []fileModuleBinding
+				if err := valueNode.Decode(&validators); err != nil {
+					return err
+				}
+				b.Validators.Validators = make([]pipeline.ModuleBinding, len(validators))
+				for i, validator := range validators {
+					b.Validators.Validators[i] = validator.toPipelineBinding()
+				}
 			default:
 				return fmt.Errorf("field %s not found in module binding", keyNode.Value)
 			}
@@ -119,6 +130,7 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
 		Retries:    b.Retries,
 		Options:    cloneOptions(b.Options),
 		References: normalizedStringMap(b.References),
+		Validators: b.Validators,
 	}
 }
 

internal/core/config/file_config_test.go

@@ -300,6 +300,61 @@ pipelines:
 	}
 }
 
+func TestParseFileConfigStageLocalValidatorOverrides(t *testing.T) {
+	cfg := parseAndApplyConfig(t, `
+version: 2
+pipelines:
+  example:
+    input: fake/input
+    chunk:
+      module: generic
+      validators: []
+    artifacts:
+      events:
+        extract:
+          module: fake/extract
+          validators:
+            - fake/validator
+            - module: fake/llm-validator
+              llm_profile: careful
+              options:
+                threshold: 0.7
+        merge:
+          module: appendorder
+          validators: []
+        normalize:
+          module: noop
+`)
+
+	profile := cfg.Pipelines["example"]
+	if !profile.Chunk.Validators.Set || len(profile.Chunk.Validators.Validators) != 0 {
+		t.Fatalf("chunk validator override = %#v, want explicit empty", profile.Chunk.Validators)
+	}
+	lane := profile.Artifacts["events"]
+	if !lane.Extract.Validators.Set {
+		t.Fatalf("extract validator override Set = false, want true")
+	}
+	validators := lane.Extract.Validators.Validators
+	if len(validators) != 2 {
+		t.Fatalf("extract validators = %#v, want two validators", validators)
+	}
+	if validators[0].Module != "fake/validator" {
+		t.Fatalf("first validator = %#v, want fake/validator", validators[0])
+	}
+	if validators[1].Module != "fake/llm-validator" || validators[1].LLMProfile != "careful" {
+		t.Fatalf("second validator = %#v, want LLM validator with profile", validators[1])
+	}
+	if validators[1].Options["threshold"] != 0.7 {
+		t.Fatalf("second validator options = %#v, want threshold", validators[1].Options)
+	}
+	if !lane.Merge.Validators.Set || len(lane.Merge.Validators.Validators) != 0 {
+		t.Fatalf("merge validator override = %#v, want explicit empty", lane.Merge.Validators)
+	}
+	if lane.Normalize.Validators.Set {
+		t.Fatalf("normalize validator override Set = true, want omitted")
+	}
+}
+
 func TestApplyFileConfigRejectsDuplicateTrimmedPipelineIDs(t *testing.T) {
 	fileCfg, err := ParseFileConfigYAML([]byte(`
 version: 2

internal/core/config/redaction.go

@@ -27,6 +27,12 @@ func cloneResolvedPipeline(in pipeline.ResolvedPipeline) pipeline.ResolvedPipeli
 	out.Chunk = cloneModuleBinding(in.Chunk)
 	out.ChunkReferences = pipeline.CloneReferenceTarget(in.ChunkReferences)
 	out.Output = cloneModuleBinding(in.Output)
+	if len(in.ValidatorChains) > 0 {
+		out.ValidatorChains = make([]pipeline.ResolvedValidatorChain, len(in.ValidatorChains))
+		for i, chain := range in.ValidatorChains {
+			out.ValidatorChains[i] = cloneResolvedValidatorChain(chain)
+		}
+	}
 	if len(in.ArtifactLanes) > 0 {
 		out.ArtifactLanes = make([]pipeline.ResolvedArtifactLane, len(in.ArtifactLanes))
 		for i, lane := range in.ArtifactLanes {
@@ -36,6 +42,20 @@ func cloneResolvedPipeline(in pipeline.ResolvedPipeline) pipeline.ResolvedPipeli
 	return out
 }
 
+func cloneResolvedValidatorChain(in pipeline.ResolvedValidatorChain) pipeline.ResolvedValidatorChain {
+	out := in
+	if len(in.Validators) > 0 {
+		out.Validators = make([]pipeline.ResolvedValidator, len(in.Validators))
+		for i, validator := range in.Validators {
+			out.Validators[i] = pipeline.ResolvedValidator{
+				Binding:        cloneModuleBinding(validator.Binding),
+				ExecutionClass: validator.ExecutionClass,
+			}
+		}
+	}
+	return out
+}
+
 func cloneResolvedArtifactLane(in pipeline.ResolvedArtifactLane) pipeline.ResolvedArtifactLane {
 	out := in
 	out.Extract = cloneModuleBinding(in.Extract)

internal/core/config/validation.go

@@ -85,7 +85,7 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro
 				return err
 			}
 			if len(lane.Validators) > 0 {
-				return fmt.Errorf("pipeline %q lane %q validators are not supported by the current raw validation runner", id, laneID)
+				return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", id, laneID)
 			}
 		}
 	}
@@ -108,6 +108,9 @@ func validateBinding(
 		}
 		return fmt.Errorf("pipeline %q %s retries must be greater than or equal to zero", pipelineID, slot)
 	}
+	if err := validateValidatorOverride(pipelineID, laneID, slot, binding.Validators); err != nil {
+		return err
+	}
 	if len(binding.References) == 0 {
 		return nil
 	}
@@ -120,6 +123,36 @@ func validateBinding(
 	return validateReferenceMapForContext(pipelineID, laneID, slot, binding.References)
 }
 
+func validateValidatorOverride(pipelineID string, laneID string, slot string, override pipeline.ValidatorOverride) error {
+	if !override.Set {
+		return nil
+	}
+	switch slot {
+	case "chunk", "extract", "merge", "normalize":
+	default:
+		return fmt.Errorf("%s validators are not supported", referenceContext(pipelineID, laneID, slot))
+	}
+	for i, validator := range override.Validators {
+		context := fmt.Sprintf("%s validators[%d]", referenceContext(pipelineID, laneID, slot), i)
+		if strings.TrimSpace(validator.Module) == "" {
+			return fmt.Errorf("%s module must not be empty", context)
+		}
+		if len(validator.References) > 0 {
+			return fmt.Errorf("%s references are not supported", context)
+		}
+		if validator.Validators.Set {
+			return fmt.Errorf("%s nested validators are not supported", context)
+		}
+		if validator.Retries != 0 {
+			return fmt.Errorf("%s retries are not supported", context)
+		}
+		if validator.LLMProfile != "" && strings.TrimSpace(validator.LLMProfile) == "" {
+			return fmt.Errorf("%s llm_profile must not be empty when set", context)
+		}
+	}
+	return nil
+}
+
 func validateReferenceMap(pipelineID string, laneID string, references map[string]string) error {
 	return validateReferenceMapForContext(pipelineID, laneID, "", references)
 }

internal/core/config/validation_test.go

@@ -341,13 +341,82 @@ func TestValidateRejectsConfiguredValidators(t *testing.T) {
 	if err == nil {
 		t.Fatal("Validate() error = nil, want configured validators error")
 	}
-	for _, want := range []string{"example", "events", "validators", "not supported"} {
+	for _, want := range []string{"example", "events", "validators", "extract.validators", "merge.validators", "normalize.validators"} {
 		if !strings.Contains(err.Error(), want) {
 			t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want)
 		}
 	}
 }
 
+func TestValidateAcceptsStageLocalValidatorOverrides(t *testing.T) {
+	cfg := validConfig()
+	profile := cfg.Pipelines["example"]
+	profile.Chunk.Validators = pipeline.ValidatorOverride{Set: true}
+	lane := profile.Artifacts["events"]
+	lane.Extract.Validators = pipeline.ValidatorOverride{
+		Set: true,
+		Validators: []pipeline.ModuleBinding{
+			pipeline.Binding("fake/validator"),
+			{Module: "fake/llm-validator", LLMProfile: "careful", Options: map[string]any{"threshold": 0.7}},
+		},
+	}
+	lane.Merge.Validators = pipeline.ValidatorOverride{Set: true}
+	profile.Artifacts["events"] = lane
+	cfg.Pipelines["example"] = profile
+
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+}
+
+func TestValidateRejectsInvalidValidatorBindings(t *testing.T) {
+	tests := []struct {
+		name    string
+		binding pipeline.ModuleBinding
+		want    string
+	}{
+		{
+			name:    "empty module",
+			binding: pipeline.ModuleBinding{},
+			want:    "module must not be empty",
+		},
+		{
+			name:    "references",
+			binding: pipeline.ModuleBinding{Module: "fake/validator", References: map[string]string{"roster": "./roster.txt"}},
+			want:    "references are not supported",
+		},
+		{
+			name:    "nested validators",
+			binding: pipeline.ModuleBinding{Module: "fake/validator", Validators: pipeline.ValidatorOverride{Set: true}},
+			want:    "nested validators are not supported",
+		},
+		{
+			name:    "retries",
+			binding: pipeline.ModuleBinding{Module: "fake/validator", Retries: 1},
+			want:    "retries are not supported",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := validConfig()
+			profile := cfg.Pipelines["example"]
+			lane := profile.Artifacts["events"]
+			lane.Extract.Validators = pipeline.ValidatorOverride{
+				Set:        true,
+				Validators: []pipeline.ModuleBinding{test.binding},
+			}
+			profile.Artifacts["events"] = lane
+			cfg.Pipelines["example"] = profile
+
+			err := cfg.Validate()
+			if err == nil || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Validate() error = %v, want %q", err, test.want)
+			}
+		})
+	}
+}
+
 func validConfig() Config {
 	cfg := Default()
 	cfg.Pipelines["example"] = pipeline.PipelineProfile{
@@ -402,6 +471,12 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module
 			Requires: []string{"normalized"},
 			Provides: []string{"validated"},
 		},
+		"fake/llm-validator": {
+			Key:      "fake/llm-validator",
+			Stage:    pipeline.StageValidate,
+			Requires: []string{"normalized"},
+			Provides: []string{"validated"},
+		},
 		"json": {
 			Key:      "json",
 			Stage:    pipeline.StageOutput,
@@ -426,6 +501,7 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module
 	mustRegisterMerger(t, mergers, specs["appendorder"])
 	mustRegisterNormalizer(t, normalizers, specs["noop"])
 	mustRegisterValidator(t, validators, specs["fake/validator"])
+	mustRegisterValidator(t, validators, specs["fake/llm-validator"])
 	mustRegisterOutput(t, outputs, specs["json"])
 
 	return pipeline.ModuleCatalog{
@@ -477,7 +553,11 @@ func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry,
 
 func mustRegisterValidator(t *testing.T, registry *pipeline.ValidatorRegistry, spec pipeline.ModuleSpec) {
 	t.Helper()
-	validatorSpec := pipeline.ValidatorSpec{Key: spec.Key, ExecutionClass: contracts.ExecutionClassDeterministic}
+	executionClass := contracts.ExecutionClassDeterministic
+	if spec.Key == "fake/llm-validator" {
+		executionClass = contracts.ExecutionClassLLMBacked
+	}
+	validatorSpec := pipeline.ValidatorSpec{Key: spec.Key, ExecutionClass: executionClass}
 	if err := registry.RegisterWithSpec(validatorSpec, func() (contracts.Validator, error) { return nil, nil }); err != nil {
 		t.Fatalf("register validator: %v", err)
 	}