diff --git a/lib/ansible/constants.py b/lib/ansible/constants.py
index 9ffa32b608..278d567baa 100644
--- a/lib/ansible/constants.py
+++ b/lib/ansible/constants.py
@@ -190,6 +190,12 @@ DEFAULT_BECOME_ASK_PASS   = get_config(p, 'privilege_escalation', 'become_ask_pa
 
 
 # PLUGINS
+
+# Modules that can optimize with_items loops into a single call.  Currently
+# these modules must (1) take a "name" or "pkg" parameter that is a list.  If
+# the module takes both, bad things could happen.
+# In the future we should probably generalize this even further
+# (mapping of param: squash field)
 DEFAULT_SQUASH_ACTIONS         = get_config(p, DEFAULTS, 'squash_actions',     'ANSIBLE_SQUASH_ACTIONS', "apt, yum, pkgng, zypper, dnf", islist=True)
 # paths
 DEFAULT_ACTION_PLUGIN_PATH     = get_config(p, DEFAULTS, 'action_plugins',     'ANSIBLE_ACTION_PLUGINS', '~/.ansible/plugins/action:/usr/share/ansible/plugins/action', ispath=True)
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 4b03dc0846..158e34680d 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -25,7 +25,7 @@ import subprocess
 import sys
 import time
 
-from ansible.compat.six import iteritems
+from ansible.compat.six import iteritems, string_types
 
 from ansible import constants as C
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleConnectionFailure
@@ -230,22 +230,29 @@ class TaskExecutor:
         (typically package management modules).
         '''
         if len(items) > 0 and self._task.action in self.SQUASH_ACTIONS:
-            final_items = []
-            name = self._task.args.pop('name', None) or self._task.args.pop('pkg', None)
-            for item in items:
-                variables['item'] = item
-                templar = Templar(loader=self._loader, shared_loader_obj=self._shared_loader_obj, variables=variables)
-                if self._task.evaluate_conditional(templar, variables):
-                    if templar._contains_vars(name):
-                        new_item = templar.template(name)
-                        final_items.append(new_item)
-                    else:
-                        final_items.append(item)
-            joined_items = ",".join(final_items)
-            self._task.args['name'] = joined_items
-            return [joined_items]
-        else:
-            return items
+            if all(isinstance(o, string_types) for o in items):
+                final_items = []
+                name = self._task.args.pop('name', None) or self._task.args.pop('pkg', None)
+                for item in items:
+                    variables['item'] = item
+                    templar = Templar(loader=self._loader, shared_loader_obj=self._shared_loader_obj, variables=variables)
+                    if self._task.evaluate_conditional(templar, variables):
+                        if templar._contains_vars(name):
+                            new_item = templar.template(name)
+                            final_items.append(new_item)
+                        else:
+                            final_items.append(item)
+                self._task.args['name'] = final_items
+                return [final_items]
+            # Right now we only optimize single entries.  In the future we
+            # could optimize more types:
+            # * lists can be squashed together
+            # * dicts could squash entries that match in all cases except the
+            #   name or pkg field.
+            # Note: we really should be checking that the name or pkg field
+            # contains a template that expands with our with_items values.
+            # If it doesn't then we may break things
+        return items
 
     def _execute(self, variables=None):
         '''