diff --git a/changelogs/fragments/skip-role-task-iterator.yml b/changelogs/fragments/skip-role-task-iterator.yml
new file mode 100644
index 00000000000..1cf6b4cbb84
--- /dev/null
+++ b/changelogs/fragments/skip-role-task-iterator.yml
@@ -0,0 +1,2 @@
+minor_changes:
+  - PlayIterator - do not return tasks from already executed roles so specific strategy plugins do not have to do the filtering of such tasks themselves
diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index 83dd5d89c31..1ef7451095e 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -440,7 +440,7 @@ class PlayIterator:
                     else:
                         state.cur_handlers_task += 1
                         if task.is_host_notified(host):
-                            break
+                            return state, task
 
             elif state.run_state == IteratingStates.COMPLETE:
                 return (state, None)
@@ -463,9 +463,15 @@ class PlayIterator:
                         and all(not h.notified_hosts for h in self.handlers)
                     )
                 ):
-                    continue
-
-                break
+                    display.debug("No handler notifications for %s, skipping." % host.name)
+                elif (
+                    (role := task._role)
+                    and role._metadata.allow_duplicates is False
+                    and host.name in self._play._get_cached_role(role)._completed
+                ):
+                    display.debug("'%s' skipped because role has already run" % task)
+                else:
+                    break
 
         return (state, task)
 
diff --git a/lib/ansible/playbook/play.py b/lib/ansible/playbook/play.py
index 877af1e0d5f..aea7ba5a806 100644
--- a/lib/ansible/playbook/play.py
+++ b/lib/ansible/playbook/play.py
@@ -19,7 +19,7 @@ from __future__ import annotations
 
 from ansible import constants as C
 from ansible import context
-from ansible.errors import AnsibleParserError, AnsibleAssertionError
+from ansible.errors import AnsibleParserError, AnsibleAssertionError, AnsibleError
 from ansible.module_utils.common.text.converters import to_native
 from ansible.module_utils.common.collections import is_sequence
 from ansible.module_utils.six import binary_type, string_types, text_type
@@ -100,6 +100,15 @@ class Play(Base, Taggable, CollectionSearch):
     def __repr__(self):
         return self.get_name()
 
+    def _get_cached_role(self, role):
+        role_path = role.get_role_path()
+        role_cache = self.role_cache[role_path]
+        try:
+            idx = role_cache.index(role)
+            return role_cache[idx]
+        except ValueError:
+            raise AnsibleError(f'Cannot locate {role.get_name()} in role cache')
+
     def _validate_hosts(self, attribute, name, value):
         # Only validate 'hosts' if a value was passed in to original data set.
         if 'hosts' in self._ds:
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index ce4ae897445..c2a3e8a618a 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -1102,13 +1102,7 @@ class StrategyBase:
         return [res]
 
     def _get_cached_role(self, task, play):
-        role_path = task._role.get_role_path()
-        role_cache = play.role_cache[role_path]
-        try:
-            idx = role_cache.index(task._role)
-            return role_cache[idx]
-        except ValueError:
-            raise AnsibleError(f'Cannot locate {task._role.get_name()} in role cache')
+        return play._get_cached_role(task._role)
 
     def get_hosts_left(self, iterator):
         ''' returns list of available hosts for this iterator by filtering out unreachables '''
diff --git a/lib/ansible/plugins/strategy/free.py b/lib/ansible/plugins/strategy/free.py
index 04c16024d9f..90840dab63b 100644
--- a/lib/ansible/plugins/strategy/free.py
+++ b/lib/ansible/plugins/strategy/free.py
@@ -172,15 +172,6 @@ class StrategyModule(StrategyBase):
                                 display.warning("Using run_once with the free strategy is not currently supported. This task will still be "
                                                 "executed for every host in the inventory list.")
 
-                        # check to see if this task should be skipped, due to it being a member of a
-                        # role which has already run (and whether that role allows duplicate execution)
-                        if not isinstance(task, Handler) and task._role:
-                            role_obj = self._get_cached_role(task, iterator._play)
-                            if role_obj.has_run(host) and task._role._metadata.allow_duplicates is False:
-                                display.debug("'%s' skipped because role has already run" % task, host=host_name)
-                                del self._blocked_hosts[host_name]
-                                continue
-
                         if task.action in C._ACTION_META:
                             if self._host_pinned:
                                 meta_task_dummy_results_count += 1
diff --git a/lib/ansible/plugins/strategy/linear.py b/lib/ansible/plugins/strategy/linear.py
index 49cfdd4bf44..c6815a38deb 100644
--- a/lib/ansible/plugins/strategy/linear.py
+++ b/lib/ansible/plugins/strategy/linear.py
@@ -130,14 +130,6 @@ class StrategyModule(StrategyBase):
                     run_once = False
                     work_to_do = True
 
-                    # check to see if this task should be skipped, due to it being a member of a
-                    # role which has already run (and whether that role allows duplicate execution)
-                    if not isinstance(task, Handler) and task._role:
-                        role_obj = self._get_cached_role(task, iterator._play)
-                        if role_obj.has_run(host) and task._role._metadata.allow_duplicates is False:
-                            display.debug("'%s' skipped because role has already run" % task)
-                            continue
-
                     display.debug("getting variables")
                     task_vars = self._variable_manager.get_vars(play=iterator._play, host=host, task=task,
                                                                 _hosts=self._hosts_cache, _hosts_all=self._hosts_cache_all)