diff --git a/lib/ansible/executor/process/result.py b/lib/ansible/executor/process/result.py
index baf7afcf5b4..68a458bd869 100644
--- a/lib/ansible/executor/process/result.py
+++ b/lib/ansible/executor/process/result.py
@@ -122,18 +122,6 @@ class ResultProcess(multiprocessing.Process):
                 elif result.is_skipped():
                     self._send_result(('host_task_skipped', result))
                 else:
-                    # if this task is notifying a handler, do it now
-                    if result._task.notify and result._result.get('changed', False):
-                        # The shared dictionary for notified handlers is a proxy, which
-                        # does not detect when sub-objects within the proxy are modified.
-                        # So, per the docs, we reassign the list so the proxy picks up and
-                        # notifies all other threads
-                        for notify in result._task.notify:
-                            if result._task._role:
-                                role_name = result._task._role.get_name()
-                                notify = "%s : %s" %(role_name, notify)
-                            self._send_result(('notify_handler', result._host, notify))
-
                     if result._task.loop:
                         # this task had a loop, and has more than one result, so
                         # loop over all of them instead of a single result
@@ -142,6 +130,20 @@ class ResultProcess(multiprocessing.Process):
                         result_items = [ result._result ]
 
                     for result_item in result_items:
+                        # if this task is notifying a handler, do it now
+                        if 'ansible_notify' in result_item and result.is_changed():
+                            # The shared dictionary for notified handlers is a proxy, which
+                            # does not detect when sub-objects within the proxy are modified.
+                            # So, per the docs, we reassign the list so the proxy picks up and
+                            # notifies all other threads
+                            for notify in result_item['ansible_notify']:
+                                if result._task._role:
+                                    role_name = result._task._role.get_name()
+                                    notify = "%s : %s" % (role_name, notify)
+                                self._send_result(('notify_handler', result._host, notify))
+                            # now remove the notify field from the results, as its no longer needed
+                            result_item.pop('ansible_notify')
+
                         if 'add_host' in result_item:
                             # this task added a new host (add_host module)
                             self._send_result(('add_host', result_item))
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index a1930e5e14d..4322310603f 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -330,6 +330,12 @@ class TaskExecutor:
         if 'ansible_facts' in result:
             variables.update(result['ansible_facts'])
 
+        # save the notification target in the result, if it was specified, as
+        # this task may be running in a loop in which case the notification
+        # may be item-specific, ie. "notify: service {{item}}"
+        if self._task.notify:
+            result['ansible_notify'] = self._task.notify
+
         # and return
         debug("attempt loop complete, returning result")
         return result
diff --git a/lib/ansible/plugins/action/normal.py b/lib/ansible/plugins/action/normal.py
index 445d8a7ae77..8e2f5c84cdf 100644
--- a/lib/ansible/plugins/action/normal.py
+++ b/lib/ansible/plugins/action/normal.py
@@ -23,7 +23,13 @@ class ActionModule(ActionBase):
 
     def run(self, tmp=None, task_vars=dict()):
 
-        #vv("REMOTE_MODULE %s %s" % (module_name, module_args), host=conn.host)
-        return self._execute_module(tmp, task_vars=task_vars)
+        results = self._execute_module(tmp, task_vars=task_vars)
 
+        # Remove special fields from the result, which can only be set
+        # internally by the executor engine. We do this only here in
+        # the 'normal' action, as other action plugins may set this.
+        for field in ('ansible_facts', 'ansible_notify'):
+            if field in results:
+                results.pop(field)
 
+        return results