Harden diagnostic handling and warning presentation

internal/framework/pipeline/producer_attempts.go

@@ -193,6 +193,10 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
 			}
 			return failedProducerAttempt(provenance), fmt.Errorf("producer failed after %d attempt(s): %w", number, err)
 		}
+		if err := validateProducerAttemptDiagnostics(output); err != nil {
+			provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed})
+			return failedProducerAttempt(provenance), err
+		}
 
 		output, err = output.clone()
 		if err != nil {
@@ -265,6 +269,18 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
 	return failedProducerAttempt(provenance), errors.New("producer attempt budget was not exhausted deterministically")
 }
 
+func validateProducerAttemptDiagnostics(output producerAttemptOutput) error {
+	if err := contracts.ValidateProducerDiagnostics(output.Diagnostics); err != nil {
+		return fmt.Errorf("producer returned invalid diagnostics: %w", err)
+	}
+	if output.Retry != nil {
+		if err := contracts.ValidateProducerDiagnostics(output.Retry.FallbackDiagnostics); err != nil {
+			return fmt.Errorf("producer returned invalid retry fallback diagnostics: %w", err)
+		}
+	}
+	return nil
+}
+
 func isImmediateProducerFailure(err error) bool {
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return true