Hardened the proposal modules to skip malformed proposals rather than hard failing the entire run

internal/framework/runner/runner.go

@@ -238,6 +238,8 @@ type sectionProposalResult struct {
 	sectionPos int
 	section    chunking.Section
 	corrected  []proposals.CorrectionProposal
+	dropped    []droppedProposalCandidate
+	rawCount   int
 	err        error
 }
 
@@ -257,6 +259,22 @@ type modulePipelineResult struct {
 	ValidatorRejected  []ValidatorRejectedChange
 }
 
+type droppedProposalCandidate struct {
+	ProposalIndex   int
+	TargetSegmentID int
+	OriginalText    string
+	CorrectedText   string
+	ReasonCode      string
+	Message         string
+}
+
+const (
+	validatorNameProposalGeneration  = "proposal_generation"
+	reasonValidatorExecutionError    = "validator_execution_error"
+	reasonValidatorMalformedResponse = "validator_malformed_response"
+	reasonValidatorMissingDecision   = "validator_missing_decision"
+)
+
 func collectSectionProposals(ctx context.Context, input collectSectionProposalsInput) (context.Context, <-chan sectionProposalResult, context.CancelFunc) {
 	results := make(chan sectionProposalResult, len(input.Sections))
 	runCtx, cancel := context.WithCancel(ctx)
@@ -271,6 +289,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI
 			go func() {
 				defer wg.Done()
 				meta := contracts.SectionMetadataFromSection(section)
+				dropped := make([]droppedProposalCandidate, 0)
 				corrected, err := input.Module.Propose(withModuleInstanceContext(runCtx, input.Spec.InstanceName), contracts.ProposalRequest{
 					ExecutionContext: contracts.ExecutionContext{
 						Config:            input.Config,
@@ -286,12 +305,25 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI
 					},
 					LLMClient:    input.ProposalClient,
 					LLMScheduler: input.ProposalScheduler,
+					OnDroppedCandidate: func(index int, targetSegmentID int, originalText string, correctedText string, confidence float64, reasonCode string, message string) {
+						_ = confidence
+						dropped = append(dropped, droppedProposalCandidate{
+							ProposalIndex:   index,
+							TargetSegmentID: targetSegmentID,
+							OriginalText:    originalText,
+							CorrectedText:   correctedText,
+							ReasonCode:      reasonCode,
+							Message:         message,
+						})
+					},
 				})
 				select {
 				case results <- sectionProposalResult{
 					sectionPos: sectionPos,
 					section:    section,
 					corrected:  corrected,
+					dropped:    dropped,
+					rawCount:   len(corrected) + len(dropped),
 					err:        err,
 				}:
 				case <-runCtx.Done():
@@ -362,7 +394,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
 			delete(pending, nextSectionToProcess)
 			sectionMeta := contracts.SectionMetadataFromSection(sectionResult.section)
 			sectionEnriched := make([]proposals.EnrichedCorrectionProposal, 0, len(sectionResult.corrected))
-			for i, corrected := range sectionResult.corrected {
+				for i, corrected := range sectionResult.corrected {
 				sectionIndex := sectionMeta.Index
 				sectionEnriched = append(sectionEnriched, proposals.EnrichedCorrectionProposal{
 					CorrectionProposal: corrected,
@@ -373,9 +405,22 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
 						SectionIndex:   &sectionIndex,
 					},
 				})
-			}
-			nextProposalIndex += len(sectionEnriched)
-			out.ProposalCount += len(sectionEnriched)
+				}
+				out.ProposalCount += sectionResult.rawCount
+				nextProposalIndex += sectionResult.rawCount
+				for _, dropped := range sectionResult.dropped {
+					out.ValidatorRejected = append(out.ValidatorRejected, ValidatorRejectedChange{
+						ValidatorName:   validatorNameProposalGeneration,
+						ProposalIndex:   dropped.ProposalIndex,
+						ModuleKey:       input.Spec.ModuleKey,
+						ModuleInstance:  input.Spec.InstanceName,
+						TargetSegmentID: dropped.TargetSegmentID,
+						OriginalText:    dropped.OriginalText,
+						CorrectedText:   dropped.CorrectedText,
+						ReasonCode:      dropped.ReasonCode,
+						Message:         dropped.Message,
+					})
+				}
 
 			validationLaunches++
 			vwg.Add(1)
@@ -508,54 +553,33 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
 			)
 		}
 		if err != nil {
-			return validateSectionCandidatesResult{
-				approved:  eligible,
-				decisions: decisions,
-				rejected:  rejected,
-			}, fmt.Errorf("validator %q failed: %w", validator.Name(), err)
-		}
-		if err := validators.EnforceDecisionCardinality(eligible, vResult.Decisions); err != nil {
-			return validateSectionCandidatesResult{
-				approved:  eligible,
-				decisions: decisions,
-				rejected:  rejected,
-			}, fmt.Errorf("validator %q cardinality failed: %w", validator.Name(), err)
-		}
-
-		nextEligible := make([]proposals.EnrichedCorrectionProposal, 0, len(eligible))
-		byIndex := make(map[int]proposals.EnrichedCorrectionProposal, len(eligible))
-		for _, p := range eligible {
-			byIndex[p.ProposalIndex] = p
-		}
-
-		for _, d := range vResult.Decisions {
-			decisions = append(decisions, ValidatorDecisionRecord{
-				ValidatorName:          validator.Name(),
-				ProposalIndex:          d.ProposalIndex,
-				Approved:               d.Approved,
-				ReasonCode:             d.ReasonCode,
-				Message:                d.Message,
-				DiagnosticArtifactPath: d.DiagnosticArtifactPath,
-			})
-			if d.Approved {
-				nextEligible = append(nextEligible, byIndex[d.ProposalIndex])
-				continue
+			for _, p := range eligible {
+				decisions = append(decisions, ValidatorDecisionRecord{
+					ValidatorName: validator.Name(),
+					ProposalIndex: p.ProposalIndex,
+					Approved:      false,
+					ReasonCode:    reasonValidatorExecutionError,
+					Message:       fmt.Sprintf("validator execution failed: %v", err),
+				})
+				rejected = append(rejected, ValidatorRejectedChange{
+					ValidatorName:   validator.Name(),
+					ProposalIndex:   p.ProposalIndex,
+					ModuleKey:       p.ModuleKey,
+					ModuleInstance:  p.ModuleInstance,
+					TargetSegmentID: p.TargetSegmentID,
+					OriginalText:    p.OriginalText,
+					CorrectedText:   p.CorrectedText,
+					ReasonCode:      reasonValidatorExecutionError,
+					Message:         fmt.Sprintf("validator execution failed: %v", err),
+				})
 			}
-			p := byIndex[d.ProposalIndex]
-			rejected = append(rejected, ValidatorRejectedChange{
-				ValidatorName:   validator.Name(),
-				ProposalIndex:   p.ProposalIndex,
-				ModuleKey:       p.ModuleKey,
-				ModuleInstance:  p.ModuleInstance,
-				TargetSegmentID: p.TargetSegmentID,
-				OriginalText:    p.OriginalText,
-				CorrectedText:   p.CorrectedText,
-				ReasonCode:      d.ReasonCode,
-				Message:         d.Message,
-			})
+			eligible = nil
+			continue
 		}
-
-		eligible = nextEligible
+		reconciled := reconcileValidatorDecisions(validator.Name(), eligible, vResult.Decisions)
+		decisions = append(decisions, reconciled.decisions...)
+		rejected = append(rejected, reconciled.rejected...)
+		eligible = reconciled.nextEligible
 	}
 
 	return validateSectionCandidatesResult{
@@ -565,6 +589,114 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
 	}, nil
 }
 
+type reconciledValidatorDecisions struct {
+	nextEligible []proposals.EnrichedCorrectionProposal
+	decisions    []ValidatorDecisionRecord
+	rejected     []ValidatorRejectedChange
+}
+
+func reconcileValidatorDecisions(validatorName string, eligible []proposals.EnrichedCorrectionProposal, in []validators.Decision) reconciledValidatorDecisions {
+	byIndex := make(map[int]proposals.EnrichedCorrectionProposal, len(eligible))
+	for _, p := range eligible {
+		byIndex[p.ProposalIndex] = p
+	}
+
+	seen := make(map[int]struct{}, len(in))
+	forcedReject := make(map[int]bool)
+	valid := make(map[int]validators.Decision, len(in))
+	for _, d := range in {
+		if _, ok := byIndex[d.ProposalIndex]; !ok {
+			continue
+		}
+		if _, dup := seen[d.ProposalIndex]; dup {
+			forcedReject[d.ProposalIndex] = true
+			continue
+		}
+		seen[d.ProposalIndex] = struct{}{}
+		valid[d.ProposalIndex] = d
+	}
+
+	nextEligible := make([]proposals.EnrichedCorrectionProposal, 0, len(eligible))
+	decisions := make([]ValidatorDecisionRecord, 0, len(eligible))
+	rejected := make([]ValidatorRejectedChange, 0, len(eligible))
+
+	for _, p := range eligible {
+		idx := p.ProposalIndex
+		if forcedReject[idx] {
+			decisions = append(decisions, ValidatorDecisionRecord{
+				ValidatorName: validatorName,
+				ProposalIndex: idx,
+				Approved:      false,
+				ReasonCode:    reasonValidatorMalformedResponse,
+				Message:       "validator returned malformed response for this proposal index",
+			})
+			rejected = append(rejected, ValidatorRejectedChange{
+				ValidatorName:   validatorName,
+				ProposalIndex:   p.ProposalIndex,
+				ModuleKey:       p.ModuleKey,
+				ModuleInstance:  p.ModuleInstance,
+				TargetSegmentID: p.TargetSegmentID,
+				OriginalText:    p.OriginalText,
+				CorrectedText:   p.CorrectedText,
+				ReasonCode:      reasonValidatorMalformedResponse,
+				Message:         "validator returned malformed response for this proposal index",
+			})
+			continue
+		}
+		d, ok := valid[idx]
+		if !ok {
+			decisions = append(decisions, ValidatorDecisionRecord{
+				ValidatorName: validatorName,
+				ProposalIndex: idx,
+				Approved:      false,
+				ReasonCode:    reasonValidatorMissingDecision,
+				Message:       "validator did not return a decision for this proposal index",
+			})
+			rejected = append(rejected, ValidatorRejectedChange{
+				ValidatorName:   validatorName,
+				ProposalIndex:   p.ProposalIndex,
+				ModuleKey:       p.ModuleKey,
+				ModuleInstance:  p.ModuleInstance,
+				TargetSegmentID: p.TargetSegmentID,
+				OriginalText:    p.OriginalText,
+				CorrectedText:   p.CorrectedText,
+				ReasonCode:      reasonValidatorMissingDecision,
+				Message:         "validator did not return a decision for this proposal index",
+			})
+			continue
+		}
+		decisions = append(decisions, ValidatorDecisionRecord{
+			ValidatorName:          validatorName,
+			ProposalIndex:          d.ProposalIndex,
+			Approved:               d.Approved,
+			ReasonCode:             d.ReasonCode,
+			Message:                d.Message,
+			DiagnosticArtifactPath: d.DiagnosticArtifactPath,
+		})
+		if d.Approved {
+			nextEligible = append(nextEligible, p)
+			continue
+		}
+		rejected = append(rejected, ValidatorRejectedChange{
+			ValidatorName:   validatorName,
+			ProposalIndex:   p.ProposalIndex,
+			ModuleKey:       p.ModuleKey,
+			ModuleInstance:  p.ModuleInstance,
+			TargetSegmentID: p.TargetSegmentID,
+			OriginalText:    p.OriginalText,
+			CorrectedText:   p.CorrectedText,
+			ReasonCode:      d.ReasonCode,
+			Message:         d.Message,
+		})
+	}
+
+	return reconciledValidatorDecisions{
+		nextEligible: nextEligible,
+		decisions:    decisions,
+		rejected:     rejected,
+	}
+}
+
 func reorderValidatorsForPipeline(in []contracts.Validator) ([]contracts.Validator, map[string]int) {
 	deterministic := make([]contracts.Validator, 0, len(in))
 	llmBacked := make([]contracts.Validator, 0, len(in))