Add validation policy configuration

2026-08-26 23:34:06 +00:00
parent 1c3da3e869
commit 9d0faabf61
14 changed files with 628 additions and 57 deletions
--- a/internal/core/config/config.go
+++ b/internal/core/config/config.go
@@ -116,6 +116,7 @@ func (c *ConcurrencyConfig) recomputeStageWorkerDefaults() {

 func clonePipelineProfile(in pipeline.PipelineProfile) pipeline.PipelineProfile {
 	out := in
+	out.ValidationPolicy = cloneValidationPolicyOverride(in.ValidationPolicy)
 	if in.StructuredOutputRepairAttempts != nil {
 		value := *in.StructuredOutputRepairAttempts
 		out.StructuredOutputRepairAttempts = &value
@@ -200,6 +201,7 @@ func cloneReferenceSource(in pipeline.ReferenceSource) pipeline.ReferenceSource

 func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding {
 	out := in
+	out.ValidationPolicy = cloneValidationPolicyOverride(in.ValidationPolicy)
 	if in.StructuredOutputRepairAttempts != nil {
 		value := *in.StructuredOutputRepairAttempts
 		out.StructuredOutputRepairAttempts = &value
@@ -212,6 +214,26 @@ func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding {
 	return out
 }

+func cloneValidationPolicyOverride(in *pipeline.ValidationPolicyOverride) *pipeline.ValidationPolicyOverride {
+	if in == nil {
+		return nil
+	}
+	out := *in
+	if in.ProducerStructuralFailure != nil {
+		value := *in.ProducerStructuralFailure
+		out.ProducerStructuralFailure = &value
+	}
+	if in.SemanticRejection != nil {
+		value := *in.SemanticRejection
+		out.SemanticRejection = &value
+	}
+	if in.ValidatorFailure != nil {
+		value := *in.ValidatorFailure
+		out.ValidatorFailure = &value
+	}
+	return &out
+}
+
 func cloneValidatorOverride(in pipeline.ValidatorOverride) pipeline.ValidatorOverride {
 	out := pipeline.ValidatorOverride{Set: in.Set}
 	if len(in.Validators) > 0 {
--- a/internal/core/config/file_config.go
+++ b/internal/core/config/file_config.go
@@ -37,6 +37,7 @@ type FilePromptKitLocalBackendConfig struct {
 type FilePipelineProfile struct {
 	LLMProfile                     *string                            `yaml:"llm_profile,omitempty"`
 	StructuredOutputRepairAttempts *int                               `yaml:"structured_output_repair_attempts,omitempty"`
+	ValidationPolicy               *pipeline.ValidationPolicyOverride `yaml:"validation_policy,omitempty"`
 	Input                          fileModuleBinding                  `yaml:"input"`
 	Chunk                          *fileModuleBinding                 `yaml:"chunk,omitempty"`
 	Artifacts                      map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"`
@@ -55,12 +56,19 @@ func (p *FilePipelineProfile) UnmarshalYAML(node *yaml.Node) error {
 	type plainFilePipelineProfile FilePipelineProfile
 	var decoded plainFilePipelineProfile
 	seen, err := decodeKnownMapping(node, &decoded, map[string]struct{}{
-		"llm_profile": {}, "structured_output_repair_attempts": {}, "input": {}, "chunk": {}, "artifacts": {}, "steps": {}, "output": {}, "references": {},
+		"llm_profile": {}, "structured_output_repair_attempts": {}, "validation_policy": {}, "input": {}, "chunk": {}, "artifacts": {}, "steps": {}, "output": {}, "references": {},
 	}, "pipeline profile")
 	if err != nil {
 		return err
 	}
 	*p = FilePipelineProfile(decoded)
+	if validationPolicyNode, ok := mappingValue(node, "validation_policy"); ok {
+		policy, err := parseValidationPolicy(validationPolicyNode, "pipeline profile")
+		if err != nil {
+			return err
+		}
+		p.ValidationPolicy = policy
+	}
 	_, p.artifactsSet = seen["artifacts"]
 	_, p.stepsSet = seen["steps"]
 	_, p.llmProfileSet = seen["llm_profile"]
@@ -154,6 +162,7 @@ type fileModuleBinding struct {
 	Module                         string
 	LLMProfile                     string
 	StructuredOutputRepairAttempts *int
+	ValidationPolicy               *pipeline.ValidationPolicyOverride
 	Retries                        int
 	Options                        map[string]any
 	References                     map[string]fileReferenceSource
@@ -250,9 +259,14 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
 		b.Module = strings.TrimSpace(module)
 		return nil
 	case yaml.MappingNode:
+		seen := make(map[string]struct{}, len(node.Content)/2)
 		for i := 0; i < len(node.Content); i += 2 {
 			keyNode := node.Content[i]
 			valueNode := node.Content[i+1]
+			if _, exists := seen[keyNode.Value]; exists {
+				return fmt.Errorf("module binding field %q is duplicated", keyNode.Value)
+			}
+			seen[keyNode.Value] = struct{}{}
 			switch keyNode.Value {
 			case "module":
 				var module string
@@ -275,6 +289,12 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
 					return err
 				}
 				b.StructuredOutputRepairAttempts = attempts
+			case "validation_policy":
+				policy, err := parseValidationPolicy(valueNode, "module binding")
+				if err != nil {
+					return err
+				}
+				b.ValidationPolicy = policy
 			case "retries":
 				var retries int
 				if err := valueNode.Decode(&retries); err != nil {
@@ -318,6 +338,7 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
 		Module:                         strings.TrimSpace(b.Module),
 		LLMProfile:                     strings.TrimSpace(b.LLMProfile),
 		StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(b.StructuredOutputRepairAttempts),
+		ValidationPolicy:               cloneValidationPolicyOverride(b.ValidationPolicy),
 		Retries:                        b.Retries,
 		Options:                        cloneOptions(b.Options),
 		References:                     fileReferenceSourcesToPipeline(b.References),
@@ -325,6 +346,51 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
 	}
 }

+func mappingValue(node *yaml.Node, key string) (*yaml.Node, bool) {
+	for i := 0; i < len(node.Content); i += 2 {
+		if node.Content[i].Value == key {
+			return node.Content[i+1], true
+		}
+	}
+	return nil, false
+}
+
+func parseValidationPolicy(node *yaml.Node, context string) (*pipeline.ValidationPolicyOverride, error) {
+	if node == nil || node.Tag == "!!null" || node.Kind != yaml.MappingNode {
+		return nil, fmt.Errorf("%s validation_policy must be an object", context)
+	}
+	policy := &pipeline.ValidationPolicyOverride{}
+	seen := make(map[string]struct{}, len(node.Content)/2)
+	for i := 0; i < len(node.Content); i += 2 {
+		key := node.Content[i].Value
+		value := node.Content[i+1]
+		if _, exists := seen[key]; exists {
+			return nil, fmt.Errorf("%s validation_policy field %q is duplicated", context, key)
+		}
+		seen[key] = struct{}{}
+		if value.Tag == "!!null" || value.Kind != yaml.ScalarNode || value.Tag != "!!str" {
+			return nil, fmt.Errorf("%s validation_policy.%s must be a string", context, key)
+		}
+		switch key {
+		case "producer_structural_failure":
+			value := pipeline.ProducerStructuralFailureAction(value.Value)
+			policy.ProducerStructuralFailure = &value
+		case "semantic_rejection":
+			value := pipeline.SemanticRejectionAction(value.Value)
+			policy.SemanticRejection = &value
+		case "validator_failure":
+			value := pipeline.ValidatorFailureAction(value.Value)
+			policy.ValidatorFailure = &value
+		default:
+			return nil, fmt.Errorf("field %s not found in %s validation_policy", key, context)
+		}
+	}
+	if err := policy.Validate(); err != nil {
+		return nil, fmt.Errorf("%s validation_policy: %w", context, err)
+	}
+	return policy, nil
+}
+
 func validateStructuredOutputRepairAttemptsNode(node *yaml.Node, context string) error {
 	if node.Kind != yaml.MappingNode {
 		return fmt.Errorf("%s must be an object", context)
@@ -573,6 +639,7 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
 			ID:                             pipelineID,
 			LLMProfile:                     llmProfile,
 			StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(filePipeline.StructuredOutputRepairAttempts),
+			ValidationPolicy:               cloneValidationPolicyOverride(filePipeline.ValidationPolicy),
 			Input:                          filePipeline.Input.toPipelineBinding(),
 			Artifacts:                      make(map[string]pipeline.ArtifactLaneProfile, len(filePipeline.Artifacts)),
 			References:                     fileReferenceSourcesToPipeline(filePipeline.References),
--- a/internal/core/config/redaction.go
+++ b/internal/core/config/redaction.go
@@ -32,6 +32,7 @@ func (e EffectiveConfig) RedactedResolvedPipelinePayload() pipeline.ResolvedPipe

 func cloneResolvedPipeline(in pipeline.ResolvedPipeline) pipeline.ResolvedPipeline {
 	out := in
+	out.ConfiguredValidationPolicy = cloneValidationPolicyOverride(in.ConfiguredValidationPolicy)
 	out.Input = redactBinding(cloneModuleBinding(in.Input))
 	out.Chunk = redactBinding(cloneModuleBinding(in.Chunk))
 	out.ChunkReferences = pipeline.CloneReferenceTarget(in.ChunkReferences)
--- a/internal/core/config/redaction_test.go
+++ b/internal/core/config/redaction_test.go
@@ -225,6 +225,39 @@ func TestRedactedResolvedPipelinePayloadHandlesTypedOptionContainers(t *testing.
 	}
 }

+func TestRedactedEffectiveConfigPayloadOwnsValidationPolicies(t *testing.T) {
+	semantic := pipeline.SemanticRejectionRejectOutput
+	validator := pipeline.ValidatorFailureFailRun
+	configured := &pipeline.ValidationPolicyOverride{SemanticRejection: &semantic, ValidatorFailure: &validator}
+	effective := EffectiveConfig{
+		Config: Config{Pipelines: map[string]pipeline.PipelineProfile{
+			"main": {ValidationPolicy: configured},
+		}},
+		ResolvedPipeline: pipeline.ResolvedPipeline{
+			ConfiguredValidationPolicy: configured,
+			ChunkValidationPolicy: pipeline.ValidationPolicy{
+				ProducerStructuralFailure: pipeline.ProducerStructuralFailureFailRun,
+				SemanticRejection:         pipeline.SemanticRejectionRejectOutput,
+				ValidatorFailure:          pipeline.ValidatorFailureFailRun,
+			},
+		},
+	}
+
+	payload := effective.RedactedSummaryPayload().(EffectiveConfig)
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(encoded), `"configured_validation_policy":{"semantic_rejection":"reject_output","validator_failure":"fail_run"}`) || !strings.Contains(string(encoded), `"chunk_validation_policy":{"producer_structural_failure":"fail_run","semantic_rejection":"reject_output","validator_failure":"fail_run"}`) {
+		t.Fatalf("redacted payload omitted validation policy: %s", encoded)
+	}
+	*payload.Config.Pipelines["main"].ValidationPolicy.SemanticRejection = pipeline.SemanticRejectionFailRun
+	*payload.ResolvedPipeline.ConfiguredValidationPolicy.ValidatorFailure = pipeline.ValidatorFailureWarnContinue
+	if *effective.Config.Pipelines["main"].ValidationPolicy.SemanticRejection != pipeline.SemanticRejectionRejectOutput || *effective.ResolvedPipeline.ConfiguredValidationPolicy.ValidatorFailure != pipeline.ValidatorFailureFailRun {
+		t.Fatal("redacted payload aliases validation policy")
+	}
+}
+
 func redactionTestBinding(name string) pipeline.ModuleBinding {
 	return pipeline.ModuleBinding{
 		Module: "safe-" + name,
--- a/internal/core/config/validation.go
+++ b/internal/core/config/validation.go
@@ -119,6 +119,11 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro
 		if err := validateStructuredOutputRepairAttempts(fmt.Sprintf("pipeline %q", id), profile.StructuredOutputRepairAttempts); err != nil {
 			return err
 		}
+		if profile.ValidationPolicy != nil {
+			if err := profile.ValidationPolicy.Validate(); err != nil {
+				return fmt.Errorf("pipeline %q validation_policy: %w", id, err)
+			}
+		}
 		if err := validateBinding(id, "", "input", profile.Input, false); err != nil {
 			return err
 		}
@@ -198,6 +203,16 @@ func validateBinding(
 	binding pipeline.ModuleBinding,
 	referencesAllowed bool,
 ) error {
+	if binding.ValidationPolicy != nil {
+		switch slot {
+		case "chunk", "extract", "merge", "normalize":
+		default:
+			return fmt.Errorf("%s validation_policy is not supported", referenceContext(pipelineID, laneID, slot))
+		}
+		if err := binding.ValidationPolicy.Validate(); err != nil {
+			return fmt.Errorf("%s validation_policy: %w", referenceContext(pipelineID, laneID, slot), err)
+		}
+	}
 	if err := validateStructuredOutputRepairAttempts(referenceContext(pipelineID, laneID, slot), binding.StructuredOutputRepairAttempts); err != nil {
 		return err
 	}
@@ -255,6 +270,9 @@ func validateValidatorOverride(pipelineID string, laneID string, slot string, ov
 		if validator.Validators.Set {
 			return fmt.Errorf("%s nested validators are not supported", context)
 		}
+		if validator.ValidationPolicy != nil {
+			return fmt.Errorf("%s validation_policy is not supported", context)
+		}
 		if validator.Retries != 0 {
 			return fmt.Errorf("%s retries are not supported", context)
 		}
--- a/internal/core/config/validation_policy_contract_test.go
+++ b/internal/core/config/validation_policy_contract_test.go
@@ -0,0 +1,97 @@
+package config
+
+import (
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestValidationPolicyFileConfigurationIsStrictAndPresenceAware(t *testing.T) {
+	const valid = `version: 4
+pipelines:
+  main:
+    validation_policy:
+      producer_structural_failure: reject_output
+      semantic_rejection: fail_run
+    input: seriatim
+    chunk:
+      module: generic
+      validation_policy:
+        validator_failure: fail_run
+    artifacts:
+      lane:
+        extract:
+          module: extract
+          validation_policy:
+            semantic_rejection: reject_output
+`
+	cfg := applyFileConfig(t, valid)
+	profile := cfg.Pipelines["main"]
+	if profile.ValidationPolicy == nil || profile.ValidationPolicy.ProducerStructuralFailure == nil || *profile.ValidationPolicy.ProducerStructuralFailure != pipeline.ProducerStructuralFailureRejectOutput || profile.ValidationPolicy.SemanticRejection == nil || *profile.ValidationPolicy.SemanticRejection != pipeline.SemanticRejectionFailRun || profile.ValidationPolicy.ValidatorFailure != nil {
+		t.Fatalf("pipeline validation policy = %#v", profile.ValidationPolicy)
+	}
+	if profile.Chunk.ValidationPolicy == nil || profile.Chunk.ValidationPolicy.ValidatorFailure == nil || *profile.Chunk.ValidationPolicy.ValidatorFailure != pipeline.ValidatorFailureFailRun {
+		t.Fatalf("chunk validation policy = %#v", profile.Chunk.ValidationPolicy)
+	}
+	lane := profile.Artifacts["lane"]
+	if lane.Extract.ValidationPolicy == nil || lane.Extract.ValidationPolicy.SemanticRejection == nil || *lane.Extract.ValidationPolicy.SemanticRejection != pipeline.SemanticRejectionRejectOutput {
+		t.Fatalf("extract validation policy = %#v", lane.Extract.ValidationPolicy)
+	}
+
+	for _, test := range []struct {
+		name string
+		yaml string
+	}{
+		{"null object", strings.Replace(valid, "validation_policy:\n      producer_structural_failure: reject_output\n      semantic_rejection: fail_run", "validation_policy: null", 1)},
+		{"null field", strings.Replace(valid, "semantic_rejection: fail_run", "semantic_rejection: null", 1)},
+		{"unknown field", strings.Replace(valid, "semantic_rejection: fail_run", "unknown: fail_run", 1)},
+		{"duplicate field", strings.Replace(valid, "semantic_rejection: fail_run", "semantic_rejection: fail_run\n      semantic_rejection: reject_output", 1)},
+		{"invalid enum", strings.Replace(valid, "semantic_rejection: fail_run", "semantic_rejection: continue", 1)},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if _, err := ParseFileConfigYAML([]byte(test.yaml)); err == nil {
+				t.Fatal("ParseFileConfigYAML() error = nil, want strict validation-policy rejection")
+			}
+		})
+	}
+}
+
+func TestValidationPolicyPlacementRules(t *testing.T) {
+	policy := &pipeline.ValidationPolicyOverride{}
+	semantic := pipeline.SemanticRejectionRejectOutput
+	policy.SemanticRejection = &semantic
+	base := pipeline.PipelineProfile{
+		ID:    "main",
+		Input: pipeline.Binding("input"),
+		Artifacts: map[string]pipeline.ArtifactLaneProfile{
+			"lane": {Extract: pipeline.Binding("extract")},
+		},
+	}
+	for _, test := range []struct {
+		name   string
+		mutate func(*pipeline.PipelineProfile)
+	}{
+		{"input", func(profile *pipeline.PipelineProfile) { profile.Input.ValidationPolicy = policy }},
+		{"output", func(profile *pipeline.PipelineProfile) {
+			profile.Output = pipeline.Binding("output")
+			profile.Output.ValidationPolicy = policy
+		}},
+		{"validator", func(profile *pipeline.PipelineProfile) {
+			lane := profile.Artifacts["lane"]
+			lane.Extract.Validators = pipeline.ValidatorOverride{Set: true, Validators: []pipeline.ModuleBinding{{Module: "validator", ValidationPolicy: policy}}}
+			profile.Artifacts["lane"] = lane
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			profile := base
+			profile.Artifacts = map[string]pipeline.ArtifactLaneProfile{"lane": base.Artifacts["lane"]}
+			test.mutate(&profile)
+			cfg := Default()
+			cfg.Pipelines = map[string]pipeline.PipelineProfile{"main": profile}
+			if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "validation_policy") {
+				t.Fatalf("Config.Validate() error = %v, want placement rejection", err)
+			}
+		})
+	}
+}
--- a/internal/framework/pipeline/prepare.go
+++ b/internal/framework/pipeline/prepare.go
@@ -406,6 +406,7 @@ func validateRegistrySet(resolved ResolvedPipeline, registries Registries) error

 func cloneResolvedPipeline(in ResolvedPipeline) ResolvedPipeline {
 	out := in
+	out.ConfiguredValidationPolicy = cloneValidationPolicyOverride(in.ConfiguredValidationPolicy)
 	out.Input = cloneModuleBinding(in.Input)
 	out.Chunk = cloneModuleBinding(in.Chunk)
 	out.Output = cloneModuleBinding(in.Output)
--- a/internal/framework/pipeline/profile.go
+++ b/internal/framework/pipeline/profile.go
@@ -24,6 +24,7 @@ type ModuleBinding struct {
 	Module                         string                     `json:"module"`
 	LLMProfile                     string                     `json:"llm_profile,omitempty"`
 	StructuredOutputRepairAttempts *int                       `json:"structured_output_repair_attempts,omitempty"`
+	ValidationPolicy               *ValidationPolicyOverride  `json:"validation_policy,omitempty"`
 	Retries                        int                        `json:"retries,omitempty"`
 	Options                        map[string]any             `json:"options,omitempty"`
 	References                     map[string]ReferenceSource `json:"references,omitempty"`
@@ -84,6 +85,7 @@ func (binding ModuleBinding) MarshalJSON() ([]byte, error) {
 		Module                         string                     `json:"module"`
 		LLMProfile                     string                     `json:"llm_profile,omitempty"`
 		StructuredOutputRepairAttempts *int                       `json:"structured_output_repair_attempts,omitempty"`
+		ValidationPolicy               *ValidationPolicyOverride  `json:"validation_policy,omitempty"`
 		Retries                        int                        `json:"retries,omitempty"`
 		Options                        map[string]any             `json:"options,omitempty"`
 		References                     map[string]ReferenceSource `json:"references,omitempty"`
@@ -93,6 +95,7 @@ func (binding ModuleBinding) MarshalJSON() ([]byte, error) {
 		Module:                         binding.Module,
 		LLMProfile:                     binding.LLMProfile,
 		StructuredOutputRepairAttempts: binding.StructuredOutputRepairAttempts,
+		ValidationPolicy:               cloneValidationPolicyOverride(binding.ValidationPolicy),
 		Retries:                        binding.Retries,
 		Options:                        binding.Options,
 		References:                     binding.References,
@@ -112,6 +115,7 @@ func (binding *ModuleBinding) UnmarshalJSON(data []byte) error {
 		Module                         string                     `json:"module"`
 		LLMProfile                     string                     `json:"llm_profile,omitempty"`
 		StructuredOutputRepairAttempts *int                       `json:"structured_output_repair_attempts,omitempty"`
+		ValidationPolicy               *ValidationPolicyOverride  `json:"validation_policy,omitempty"`
 		Retries                        int                        `json:"retries,omitempty"`
 		Options                        map[string]any             `json:"options,omitempty"`
 		References                     map[string]ReferenceSource `json:"references,omitempty"`
@@ -125,6 +129,7 @@ func (binding *ModuleBinding) UnmarshalJSON(data []byte) error {
 		Module:                         decoded.Module,
 		LLMProfile:                     decoded.LLMProfile,
 		StructuredOutputRepairAttempts: decoded.StructuredOutputRepairAttempts,
+		ValidationPolicy:               cloneValidationPolicyOverride(decoded.ValidationPolicy),
 		Retries:                        decoded.Retries,
 		Options:                        decoded.Options,
 		References:                     decoded.References,
@@ -158,6 +163,7 @@ type PipelineProfile struct {
 	ID                             string                         `json:"id"`
 	LLMProfile                     string                         `json:"llm_profile,omitempty"`
 	StructuredOutputRepairAttempts *int                           `json:"structured_output_repair_attempts,omitempty"`
+	ValidationPolicy               *ValidationPolicyOverride      `json:"validation_policy,omitempty"`
 	Input                          ModuleBinding                  `json:"input"`
 	Chunk                          ModuleBinding                  `json:"chunk,omitempty"`
 	Artifacts                      map[string]ArtifactLaneProfile `json:"artifacts"`
@@ -198,23 +204,26 @@ type ResolvedReferenceTarget struct {
 }

 type ResolvedArtifactLane struct {
-	StepID                  string
-	ID                      string
-	ArtifactKind            contracts.ArtifactKind `json:"artifact_kind,omitempty"`
-	ArtifactSchemaID        string                 `json:"artifact_schema_id,omitempty"`
-	ArtifactSchemaName      string                 `json:"artifact_schema_name,omitempty"`
-	ArtifactSchemaVersion   string                 `json:"artifact_schema_version,omitempty"`
-	ArtifactSchemaDigest    string                 `json:"artifact_schema_digest,omitempty"`
-	Extract                 ModuleBinding
-	ExtractExecutionClass   contracts.ExecutionClass `json:"extract_execution_class"`
-	Merge                   ModuleBinding
-	MergeExecutionClass     contracts.ExecutionClass `json:"merge_execution_class"`
-	Normalize               ModuleBinding
-	NormalizeExecutionClass contracts.ExecutionClass `json:"normalize_execution_class"`
-	Validators              []ModuleBinding
-	ExtractReferences       ResolvedReferenceTarget `json:"extract_references"`
-	MergeReferences         ResolvedReferenceTarget `json:"merge_references"`
-	NormalizeReferences     ResolvedReferenceTarget `json:"normalize_references"`
+	StepID                    string
+	ID                        string
+	ArtifactKind              contracts.ArtifactKind `json:"artifact_kind,omitempty"`
+	ArtifactSchemaID          string                 `json:"artifact_schema_id,omitempty"`
+	ArtifactSchemaName        string                 `json:"artifact_schema_name,omitempty"`
+	ArtifactSchemaVersion     string                 `json:"artifact_schema_version,omitempty"`
+	ArtifactSchemaDigest      string                 `json:"artifact_schema_digest,omitempty"`
+	Extract                   ModuleBinding
+	ExtractExecutionClass     contracts.ExecutionClass `json:"extract_execution_class"`
+	ExtractValidationPolicy   ValidationPolicy         `json:"extract_validation_policy"`
+	Merge                     ModuleBinding
+	MergeExecutionClass       contracts.ExecutionClass `json:"merge_execution_class"`
+	MergeValidationPolicy     ValidationPolicy         `json:"merge_validation_policy"`
+	Normalize                 ModuleBinding
+	NormalizeExecutionClass   contracts.ExecutionClass `json:"normalize_execution_class"`
+	NormalizeValidationPolicy ValidationPolicy         `json:"normalize_validation_policy"`
+	Validators                []ModuleBinding
+	ExtractReferences         ResolvedReferenceTarget `json:"extract_references"`
+	MergeReferences           ResolvedReferenceTarget `json:"merge_references"`
+	NormalizeReferences       ResolvedReferenceTarget `json:"normalize_references"`
 }

 type ResolvedPipelineStep struct {
@@ -237,17 +246,19 @@ type ResolvedValidator struct {
 }

 type ResolvedPipeline struct {
-	ID                   string
-	Digest               string
-	Input                ModuleBinding
-	InputExecutionClass  contracts.ExecutionClass `json:"input_execution_class"`
-	Chunk                ModuleBinding
-	ChunkExecutionClass  contracts.ExecutionClass `json:"chunk_execution_class"`
-	ChunkReferences      ResolvedReferenceTarget  `json:"chunk_references"`
-	Steps                []ResolvedPipelineStep
-	ValidatorChains      []ResolvedValidatorChain `json:"validator_chains"`
-	Output               ModuleBinding
-	OutputExecutionClass contracts.ExecutionClass `json:"output_execution_class"`
+	ID                         string
+	Digest                     string
+	ConfiguredValidationPolicy *ValidationPolicyOverride `json:"configured_validation_policy,omitempty"`
+	Input                      ModuleBinding
+	InputExecutionClass        contracts.ExecutionClass `json:"input_execution_class"`
+	Chunk                      ModuleBinding
+	ChunkExecutionClass        contracts.ExecutionClass `json:"chunk_execution_class"`
+	ChunkValidationPolicy      ValidationPolicy         `json:"chunk_validation_policy"`
+	ChunkReferences            ResolvedReferenceTarget  `json:"chunk_references"`
+	Steps                      []ResolvedPipelineStep
+	ValidatorChains            []ResolvedValidatorChain `json:"validator_chains"`
+	Output                     ModuleBinding
+	OutputExecutionClass       contracts.ExecutionClass `json:"output_execution_class"`
 }

 // AllArtifactLanes returns lanes in deterministic step order for read-only
@@ -414,13 +425,14 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
 		return ResolvedPipeline{}, err
 	}
 	resolved := ResolvedPipeline{
-		ID:                  pipelineID,
-		Input:               input,
-		InputExecutionClass: inputModuleSpec.ExecutionClass,
-		Chunk:               chunk,
-		ChunkExecutionClass: chunkSpec.ExecutionClass,
-		ChunkReferences:     referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
-		Output:              output,
+		ID:                         pipelineID,
+		ConfiguredValidationPolicy: cloneValidationPolicyOverride(profile.ValidationPolicy),
+		Input:                      input,
+		InputExecutionClass:        inputModuleSpec.ExecutionClass,
+		Chunk:                      chunk,
+		ChunkExecutionClass:        chunkSpec.ExecutionClass,
+		ChunkReferences:            referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
+		Output:                     output,
 	}
 	chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, chunk.Validators, "", nil, catalog)
 	if err != nil {
@@ -493,6 +505,9 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
 	if err := applyEffectiveStructuredOutputRepairAttempts(&resolved, profile.StructuredOutputRepairAttempts); err != nil {
 		return ResolvedPipeline{}, err
 	}
+	if err := applyEffectiveValidationPolicies(&resolved, profile.ValidationPolicy); err != nil {
+		return ResolvedPipeline{}, err
+	}
 	if err := validateResolvedOptions(resolved, catalog, configuredLaneIDs); err != nil {
 		return ResolvedPipeline{}, err
 	}
@@ -1328,6 +1343,7 @@ func resolveBinding(binding ModuleBinding, defaultModule string, referenceSlotLa
 		Module:                         module,
 		LLMProfile:                     llmProfile,
 		StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts),
+		ValidationPolicy:               cloneValidationPolicyOverride(binding.ValidationPolicy),
 		Retries:                        binding.Retries,
 		Options:                        cloneOptions(binding.Options),
 		References:                     references,
@@ -1456,6 +1472,72 @@ func applyEffectiveStructuredOutputRepairAttempts(resolved *ResolvedPipeline, pi
 	return nil
 }

+func applyEffectiveValidationPolicies(resolved *ResolvedPipeline, pipelinePolicy *ValidationPolicyOverride) error {
+	apply := func(stage ModuleStage, laneID string, binding ModuleBinding, executionClass contracts.ExecutionClass) (ValidationPolicy, error) {
+		binding.ValidationPolicy = cloneValidationPolicyOverride(binding.ValidationPolicy)
+		if binding.ValidationPolicy != nil {
+			if err := binding.ValidationPolicy.Validate(); err != nil {
+				return ValidationPolicy{}, fmt.Errorf("pipeline %q %s validation_policy: %w", resolved.ID, stage, err)
+			}
+			if executionClass == contracts.ExecutionClassDeterministic && binding.ValidationPolicy.ProducerStructuralFailure != nil {
+				if laneID == "" {
+					return ValidationPolicy{}, fmt.Errorf("pipeline %q %s %q assigns producer_structural_failure to deterministic module", resolved.ID, stage, binding.Module)
+				}
+				return ValidationPolicy{}, fmt.Errorf("pipeline %q lane %q %s %q assigns producer_structural_failure to deterministic module", resolved.ID, laneID, stage, binding.Module)
+			}
+		}
+		return ResolveValidationPolicy(binding.ValidationPolicy, pipelinePolicy), nil
+	}
+
+	if pipelinePolicy != nil {
+		if err := pipelinePolicy.Validate(); err != nil {
+			return fmt.Errorf("pipeline %q validation_policy: %w", resolved.ID, err)
+		}
+	}
+	if resolved.Input.ValidationPolicy != nil {
+		return fmt.Errorf("pipeline %q input validation_policy is not supported", resolved.ID)
+	}
+	if resolved.Output.ValidationPolicy != nil {
+		return fmt.Errorf("pipeline %q output validation_policy is not supported", resolved.ID)
+	}
+	for _, chain := range resolved.ValidatorChains {
+		for _, validator := range chain.Validators {
+			if validator.Binding.ValidationPolicy != nil {
+				if chain.LaneID == "" {
+					return fmt.Errorf("pipeline %q %s validator %q validation_policy is not supported", resolved.ID, chain.Stage, validator.Binding.Module)
+				}
+				return fmt.Errorf("pipeline %q lane %q %s validator %q validation_policy is not supported", resolved.ID, chain.LaneID, chain.Stage, validator.Binding.Module)
+			}
+		}
+	}
+	policy, err := apply(StageChunk, "", resolved.Chunk, resolved.ChunkExecutionClass)
+	if err != nil {
+		return err
+	}
+	resolved.ChunkValidationPolicy = policy
+	for stepIndex := range resolved.Steps {
+		for laneIndex := range resolved.Steps[stepIndex].ArtifactLanes {
+			lane := &resolved.Steps[stepIndex].ArtifactLanes[laneIndex]
+			policy, err = apply(StageExtract, lane.ID, lane.Extract, lane.ExtractExecutionClass)
+			if err != nil {
+				return err
+			}
+			lane.ExtractValidationPolicy = policy
+			policy, err = apply(StageMerge, lane.ID, lane.Merge, lane.MergeExecutionClass)
+			if err != nil {
+				return err
+			}
+			lane.MergeValidationPolicy = policy
+			policy, err = apply(StageNormalize, lane.ID, lane.Normalize, lane.NormalizeExecutionClass)
+			if err != nil {
+				return err
+			}
+			lane.NormalizeValidationPolicy = policy
+		}
+	}
+	return nil
+}
+
 func resolveBindings(bindings []ModuleBinding, defaultModule string, referenceSlotLabel string) ([]ModuleBinding, error) {
 	if len(bindings) == 0 {
 		return nil, nil
@@ -1549,27 +1631,31 @@ func selectedArtifactLanes(pipelineID string, artifacts map[string]ArtifactLaneP

 func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
 	withoutDigest := struct {
-		ID                   string
-		Input                ModuleBinding
-		InputExecutionClass  contracts.ExecutionClass
-		Chunk                ModuleBinding
-		ChunkExecutionClass  contracts.ExecutionClass
-		ChunkReferences      ResolvedReferenceTarget
-		Steps                []ResolvedPipelineStep
-		ValidatorChains      []ResolvedValidatorChain
-		Output               ModuleBinding
-		OutputExecutionClass contracts.ExecutionClass
+		ID                         string
+		ConfiguredValidationPolicy *ValidationPolicyOverride
+		Input                      ModuleBinding
+		InputExecutionClass        contracts.ExecutionClass
+		Chunk                      ModuleBinding
+		ChunkExecutionClass        contracts.ExecutionClass
+		ChunkValidationPolicy      ValidationPolicy
+		ChunkReferences            ResolvedReferenceTarget
+		Steps                      []ResolvedPipelineStep
+		ValidatorChains            []ResolvedValidatorChain
+		Output                     ModuleBinding
+		OutputExecutionClass       contracts.ExecutionClass
 	}{
-		ID:                   resolved.ID,
-		Input:                resolved.Input,
-		InputExecutionClass:  resolved.InputExecutionClass,
-		Chunk:                resolved.Chunk,
-		ChunkExecutionClass:  resolved.ChunkExecutionClass,
-		ChunkReferences:      resolved.ChunkReferences,
-		Steps:                resolved.Steps,
-		ValidatorChains:      resolved.ValidatorChains,
-		Output:               resolved.Output,
-		OutputExecutionClass: resolved.OutputExecutionClass,
+		ID:                         resolved.ID,
+		ConfiguredValidationPolicy: cloneValidationPolicyOverride(resolved.ConfiguredValidationPolicy),
+		ChunkValidationPolicy:      resolved.ChunkValidationPolicy,
+		Input:                      resolved.Input,
+		InputExecutionClass:        resolved.InputExecutionClass,
+		Chunk:                      resolved.Chunk,
+		ChunkExecutionClass:        resolved.ChunkExecutionClass,
+		ChunkReferences:            resolved.ChunkReferences,
+		Steps:                      resolved.Steps,
+		ValidatorChains:            resolved.ValidatorChains,
+		Output:                     resolved.Output,
+		OutputExecutionClass:       resolved.OutputExecutionClass,
 	}
 	encoded, err := json.Marshal(withoutDigest)
 	if err != nil {
--- a/internal/framework/pipeline/validation_policy.go
+++ b/internal/framework/pipeline/validation_policy.go
@@ -0,0 +1,115 @@
+package pipeline
+
+import "fmt"
+
+type ProducerStructuralFailureAction string
+
+const (
+	ProducerStructuralFailureFailRun      ProducerStructuralFailureAction = "fail_run"
+	ProducerStructuralFailureRejectOutput ProducerStructuralFailureAction = "reject_output"
+)
+
+type SemanticRejectionAction string
+
+const (
+	SemanticRejectionFailRun      SemanticRejectionAction = "fail_run"
+	SemanticRejectionRejectOutput SemanticRejectionAction = "reject_output"
+)
+
+type ValidatorFailureAction string
+
+const (
+	ValidatorFailureWarnContinue ValidatorFailureAction = "warn_continue"
+	ValidatorFailureFailRun      ValidatorFailureAction = "fail_run"
+)
+
+// ValidationPolicyOverride records only explicitly configured terminal-policy
+// values. Nil fields inherit independently from the pipeline or application.
+type ValidationPolicyOverride struct {
+	ProducerStructuralFailure *ProducerStructuralFailureAction `json:"producer_structural_failure,omitempty"`
+	SemanticRejection         *SemanticRejectionAction         `json:"semantic_rejection,omitempty"`
+	ValidatorFailure          *ValidatorFailureAction          `json:"validator_failure,omitempty"`
+}
+
+// ValidationPolicy is the concrete terminal policy retained by each resolved
+// producer. It contains no inherited pointers.
+type ValidationPolicy struct {
+	ProducerStructuralFailure ProducerStructuralFailureAction `json:"producer_structural_failure"`
+	SemanticRejection         SemanticRejectionAction         `json:"semantic_rejection"`
+	ValidatorFailure          ValidatorFailureAction          `json:"validator_failure"`
+}
+
+func DefaultValidationPolicy() ValidationPolicy {
+	return ValidationPolicy{
+		ProducerStructuralFailure: ProducerStructuralFailureFailRun,
+		SemanticRejection:         SemanticRejectionFailRun,
+		ValidatorFailure:          ValidatorFailureWarnContinue,
+	}
+}
+
+func (override ValidationPolicyOverride) Validate() error {
+	if override.ProducerStructuralFailure != nil {
+		switch *override.ProducerStructuralFailure {
+		case ProducerStructuralFailureFailRun, ProducerStructuralFailureRejectOutput:
+		default:
+			return fmt.Errorf("producer_structural_failure must be fail_run or reject_output")
+		}
+	}
+	if override.SemanticRejection != nil {
+		switch *override.SemanticRejection {
+		case SemanticRejectionFailRun, SemanticRejectionRejectOutput:
+		default:
+			return fmt.Errorf("semantic_rejection must be fail_run or reject_output")
+		}
+	}
+	if override.ValidatorFailure != nil {
+		switch *override.ValidatorFailure {
+		case ValidatorFailureWarnContinue, ValidatorFailureFailRun:
+		default:
+			return fmt.Errorf("validator_failure must be warn_continue or fail_run")
+		}
+	}
+	return nil
+}
+
+func ResolveValidationPolicy(binding, pipeline *ValidationPolicyOverride) ValidationPolicy {
+	resolved := DefaultValidationPolicy()
+	applyValidationPolicyOverride(&resolved, pipeline)
+	applyValidationPolicyOverride(&resolved, binding)
+	return resolved
+}
+
+func cloneValidationPolicyOverride(override *ValidationPolicyOverride) *ValidationPolicyOverride {
+	if override == nil {
+		return nil
+	}
+	out := *override
+	if override.ProducerStructuralFailure != nil {
+		value := *override.ProducerStructuralFailure
+		out.ProducerStructuralFailure = &value
+	}
+	if override.SemanticRejection != nil {
+		value := *override.SemanticRejection
+		out.SemanticRejection = &value
+	}
+	if override.ValidatorFailure != nil {
+		value := *override.ValidatorFailure
+		out.ValidatorFailure = &value
+	}
+	return &out
+}
+
+func applyValidationPolicyOverride(policy *ValidationPolicy, override *ValidationPolicyOverride) {
+	if override == nil {
+		return
+	}
+	if override.ProducerStructuralFailure != nil {
+		policy.ProducerStructuralFailure = *override.ProducerStructuralFailure
+	}
+	if override.SemanticRejection != nil {
+		policy.SemanticRejection = *override.SemanticRejection
+	}
+	if override.ValidatorFailure != nil {
+		policy.ValidatorFailure = *override.ValidatorFailure
+	}
+}
--- a/internal/framework/pipeline/validation_policy_test.go
+++ b/internal/framework/pipeline/validation_policy_test.go
@@ -0,0 +1,103 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func TestResolvePipelineAppliesValidationPolicyFieldByField(t *testing.T) {
+	profile := baselineProfile()
+	profile.ValidationPolicy = validationPolicyOverride(
+		ProducerStructuralFailureRejectOutput,
+		SemanticRejectionRejectOutput,
+		ValidatorFailureFailRun,
+	)
+	profile.Chunk.ValidationPolicy = validationPolicyOverride("", "", ValidatorFailureWarnContinue)
+	lane := profile.Artifacts["events"]
+	lane.Extract.ValidationPolicy = validationPolicyOverride("", SemanticRejectionFailRun, "")
+	lane.Merge.ValidationPolicy = validationPolicyOverride("", "", ValidatorFailureWarnContinue)
+	profile.Artifacts["events"] = lane
+
+	resolved, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t))
+	if err != nil {
+		t.Fatalf("ResolvePipeline() error = %v", err)
+	}
+	if got, want := resolved.ChunkValidationPolicy, (ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureRejectOutput, SemanticRejection: SemanticRejectionRejectOutput, ValidatorFailure: ValidatorFailureWarnContinue}); got != want {
+		t.Fatalf("chunk validation policy = %#v, want %#v", got, want)
+	}
+	resolvedLane := resolved.Steps[0].ArtifactLanes[0]
+	if got, want := resolvedLane.ExtractValidationPolicy, (ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureRejectOutput, SemanticRejection: SemanticRejectionFailRun, ValidatorFailure: ValidatorFailureFailRun}); got != want {
+		t.Fatalf("extract validation policy = %#v, want %#v", got, want)
+	}
+	if got, want := resolvedLane.MergeValidationPolicy, (ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureRejectOutput, SemanticRejection: SemanticRejectionRejectOutput, ValidatorFailure: ValidatorFailureWarnContinue}); got != want {
+		t.Fatalf("merge validation policy = %#v, want %#v", got, want)
+	}
+	if got, want := resolvedLane.NormalizeValidationPolicy, (ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureRejectOutput, SemanticRejection: SemanticRejectionRejectOutput, ValidatorFailure: ValidatorFailureFailRun}); got != want {
+		t.Fatalf("normalize validation policy = %#v, want %#v", got, want)
+	}
+
+	*profile.ValidationPolicy.SemanticRejection = SemanticRejectionFailRun
+	if got := resolved.ConfiguredValidationPolicy.SemanticRejection; got == nil || *got != SemanticRejectionRejectOutput {
+		t.Fatalf("resolved configured policy aliases profile: %#v", resolved.ConfiguredValidationPolicy)
+	}
+}
+
+func TestResolvePipelineRejectsStructuralPolicyOverrideOnDeterministicProducer(t *testing.T) {
+	profile := baselineProfile()
+	profile.Chunk.ValidationPolicy = validationPolicyOverride(ProducerStructuralFailureRejectOutput, "", "")
+	_, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t))
+	if err == nil || !strings.Contains(err.Error(), "producer_structural_failure") || !strings.Contains(err.Error(), "deterministic") {
+		t.Fatalf("ResolvePipeline() error = %v, want deterministic structural-policy rejection", err)
+	}
+}
+
+func TestValidationPolicyChangesResolvedDigestAndRoundTripsBindings(t *testing.T) {
+	left := baselineProfile()
+	right := baselineProfile()
+	right.ValidationPolicy = validationPolicyOverride("", SemanticRejectionRejectOutput, "")
+
+	leftResolved, err := ResolvePipeline(left, ResolveOptions{}, newProfileCatalog(t))
+	if err != nil {
+		t.Fatalf("ResolvePipeline(left) error = %v", err)
+	}
+	rightResolved, err := ResolvePipeline(right, ResolveOptions{}, newProfileCatalog(t))
+	if err != nil {
+		t.Fatalf("ResolvePipeline(right) error = %v", err)
+	}
+	if leftResolved.Digest == rightResolved.Digest {
+		t.Fatal("validation policy change did not change resolved digest")
+	}
+
+	binding := ModuleBinding{Module: "producer", ValidationPolicy: validationPolicyOverride(ProducerStructuralFailureRejectOutput, SemanticRejectionRejectOutput, ValidatorFailureFailRun)}
+	encoded, err := json.Marshal(binding)
+	if err != nil {
+		t.Fatalf("marshal binding: %v", err)
+	}
+	var decoded ModuleBinding
+	if err := json.Unmarshal(encoded, &decoded); err != nil {
+		t.Fatalf("unmarshal binding: %v", err)
+	}
+	if decoded.ValidationPolicy == nil || *decoded.ValidationPolicy.ProducerStructuralFailure != ProducerStructuralFailureRejectOutput || *decoded.ValidationPolicy.SemanticRejection != SemanticRejectionRejectOutput || *decoded.ValidationPolicy.ValidatorFailure != ValidatorFailureFailRun {
+		t.Fatalf("round-tripped validation policy = %#v", decoded.ValidationPolicy)
+	}
+	cloned := cloneModuleBinding(binding)
+	*binding.ValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
+	if got := *cloned.ValidationPolicy.ValidatorFailure; got != ValidatorFailureFailRun {
+		t.Fatalf("cloned validation policy aliases binding: %q", got)
+	}
+}
+
+func validationPolicyOverride(structural ProducerStructuralFailureAction, semantic SemanticRejectionAction, validator ValidatorFailureAction) *ValidationPolicyOverride {
+	policy := &ValidationPolicyOverride{}
+	if structural != "" {
+		policy.ProducerStructuralFailure = &structural
+	}
+	if semantic != "" {
+		policy.SemanticRejection = &semantic
+	}
+	if validator != "" {
+		policy.ValidatorFailure = &validator
+	}
+	return policy
+}
--- a/internal/framework/pipeline/validator_chain_registry.go
+++ b/internal/framework/pipeline/validator_chain_registry.go
@@ -95,6 +95,7 @@ func cloneModuleBinding(binding ModuleBinding) ModuleBinding {
 	binding.Module = strings.TrimSpace(binding.Module)
 	binding.LLMProfile = strings.TrimSpace(binding.LLMProfile)
 	binding.StructuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts)
+	binding.ValidationPolicy = cloneValidationPolicyOverride(binding.ValidationPolicy)
 	binding.Options = cloneOptions(binding.Options)
 	if len(binding.References) > 0 {
 		references := make(map[string]ReferenceSource, len(binding.References))