Merging upstream version 2.16.5.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

21 files changed, 284 insertions, 362 deletions

diff --git a/lib/ansible/playbook/__init__.py b/lib/ansible/playbook/__init__.py
index 0ab2271..52b2ee7 100644
--- a/lib/ansible/playbook/__init__.py
+++ b/lib/ansible/playbook/__init__.py
@@ -23,7 +23,7 @@ import os
 
 from ansible import constants as C
 from ansible.errors import AnsibleParserError
-from ansible.module_utils._text import to_text, to_native
+from ansible.module_utils.common.text.converters import to_text, to_native
 from ansible.playbook.play import Play
 from ansible.playbook.playbook_include import PlaybookInclude
 from ansible.plugins.loader import add_all_plugin_dirs
diff --git a/lib/ansible/playbook/attribute.py b/lib/ansible/playbook/attribute.py
index 692aa9a..73e73ab 100644
--- a/lib/ansible/playbook/attribute.py
+++ b/lib/ansible/playbook/attribute.py
@@ -19,8 +19,6 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
-from copy import copy, deepcopy
-
 from ansible.utils.sentinel import Sentinel
 
 _CONTAINERS = frozenset(('list', 'dict', 'set'))
diff --git a/lib/ansible/playbook/base.py b/lib/ansible/playbook/base.py
index c772df1..81ce502 100644
--- a/lib/ansible/playbook/base.py
+++ b/lib/ansible/playbook/base.py
@@ -19,7 +19,7 @@ from ansible import context
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleAssertionError
 from ansible.module_utils.six import string_types
 from ansible.module_utils.parsing.convert_bool import boolean
-from ansible.module_utils._text import to_text, to_native
+from ansible.module_utils.common.text.converters import to_text, to_native
 from ansible.parsing.dataloader import DataLoader
 from ansible.playbook.attribute import Attribute, FieldAttribute, ConnectionFieldAttribute, NonInheritableFieldAttribute
 from ansible.plugins.loader import module_loader, action_loader
@@ -486,6 +486,8 @@ class FieldAttributeBase:
             if not isinstance(value, attribute.class_type):
                 raise TypeError("%s is not a valid %s (got a %s instead)" % (name, attribute.class_type, type(value)))
             value.post_validate(templar=templar)
+        else:
+            raise AnsibleAssertionError(f"Unknown value for attribute.isa: {attribute.isa}")
         return value
 
     def set_to_context(self, name):
@@ -588,6 +590,13 @@ class FieldAttributeBase:
                 _validate_variable_keys(ds)
                 return combine_vars(self.vars, ds)
             elif isinstance(ds, list):
+                display.deprecated(
+                    (
+                        'Specifying a list of dictionaries for vars is deprecated in favor of '
+                        'specifying a dictionary.'
+                    ),
+                    version='2.18'
+                )
                 all_vars = self.vars
                 for item in ds:
                     if not isinstance(item, dict):
@@ -600,7 +609,7 @@ class FieldAttributeBase:
             else:
                 raise ValueError
         except ValueError as e:
-            raise AnsibleParserError("Vars in a %s must be specified as a dictionary, or a list of dictionaries" % self.__class__.__name__,
+            raise AnsibleParserError("Vars in a %s must be specified as a dictionary" % self.__class__.__name__,
                                      obj=ds, orig_exc=e)
         except TypeError as e:
             raise AnsibleParserError("Invalid variable name in vars specified for %s: %s" % (self.__class__.__name__, e), obj=ds, orig_exc=e)
@@ -628,7 +637,7 @@ class FieldAttributeBase:
         else:
             combined = value + new_value
 
-        return [i for i, _ in itertools.groupby(combined) if i is not None]
+        return [i for i, dummy in itertools.groupby(combined) if i is not None]
 
     def dump_attrs(self):
         '''
@@ -722,7 +731,7 @@ class Base(FieldAttributeBase):
 
     # flags and misc. settings
     environment = FieldAttribute(isa='list', extend=True, prepend=True)
-    no_log = FieldAttribute(isa='bool')
+    no_log = FieldAttribute(isa='bool', default=C.DEFAULT_NO_LOG)
     run_once = FieldAttribute(isa='bool')
     ignore_errors = FieldAttribute(isa='bool')
     ignore_unreachable = FieldAttribute(isa='bool')
diff --git a/lib/ansible/playbook/block.py b/lib/ansible/playbook/block.py
index fabaf7f..e585fb7 100644
--- a/lib/ansible/playbook/block.py
+++ b/lib/ansible/playbook/block.py
@@ -21,28 +21,25 @@ __metaclass__ = type
 
 import ansible.constants as C
 from ansible.errors import AnsibleParserError
-from ansible.playbook.attribute import FieldAttribute, NonInheritableFieldAttribute
+from ansible.playbook.attribute import NonInheritableFieldAttribute
 from ansible.playbook.base import Base
 from ansible.playbook.conditional import Conditional
 from ansible.playbook.collectionsearch import CollectionSearch
+from ansible.playbook.delegatable import Delegatable
 from ansible.playbook.helpers import load_list_of_tasks
+from ansible.playbook.notifiable import Notifiable
 from ansible.playbook.role import Role
 from ansible.playbook.taggable import Taggable
 from ansible.utils.sentinel import Sentinel
 
 
-class Block(Base, Conditional, CollectionSearch, Taggable):
+class Block(Base, Conditional, CollectionSearch, Taggable, Notifiable, Delegatable):
 
     # main block fields containing the task lists
     block = NonInheritableFieldAttribute(isa='list', default=list)
     rescue = NonInheritableFieldAttribute(isa='list', default=list)
     always = NonInheritableFieldAttribute(isa='list', default=list)
 
-    # other fields for task compat
-    notify = FieldAttribute(isa='list')
-    delegate_to = FieldAttribute(isa='string')
-    delegate_facts = FieldAttribute(isa='bool')
-
     # for future consideration? this would be functionally
     # similar to the 'else' clause for exceptions
     # otherwise = FieldAttribute(isa='list')
@@ -380,7 +377,6 @@ class Block(Base, Conditional, CollectionSearch, Taggable):
                     if filtered_block.has_tasks():
                         tmp_list.append(filtered_block)
                 elif ((task.action in C._ACTION_META and task.implicit) or
-                        (task.action in C._ACTION_INCLUDE and task.evaluate_tags([], self._play.skip_tags, all_vars=all_vars)) or
                         task.evaluate_tags(self._play.only_tags, self._play.skip_tags, all_vars=all_vars)):
                     tmp_list.append(task)
             return tmp_list
diff --git a/lib/ansible/playbook/conditional.py b/lib/ansible/playbook/conditional.py
index d994f8f..449b4a9 100644
--- a/lib/ansible/playbook/conditional.py
+++ b/lib/ansible/playbook/conditional.py
@@ -19,28 +19,18 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
-import ast
-import re
+import typing as t
 
-from jinja2.compiler import generate
-from jinja2.exceptions import UndefinedError
-
-from ansible import constants as C
 from ansible.errors import AnsibleError, AnsibleUndefinedVariable, AnsibleTemplateError
-from ansible.module_utils.six import text_type
-from ansible.module_utils._text import to_native, to_text
+from ansible.module_utils.common.text.converters import to_native
 from ansible.playbook.attribute import FieldAttribute
+from ansible.template import Templar
 from ansible.utils.display import Display
 
 display = Display()
 
-DEFINED_REGEX = re.compile(r'(hostvars\[.+\]|[\w_]+)\s+(not\s+is|is|is\s+not)\s+(defined|undefined)')
-LOOKUP_REGEX = re.compile(r'lookup\s*\(')
-VALID_VAR_REGEX = re.compile("^[_A-Za-z][_a-zA-Z0-9]*$")
-
 
 class Conditional:
-
     '''
     This is a mix-in class, to be used with Base to allow the object
     to be run conditionally when a condition is met or skipped.
@@ -57,166 +47,69 @@ class Conditional:
                 raise AnsibleError("a loader must be specified when using Conditional() directly")
             else:
                 self._loader = loader
-        super(Conditional, self).__init__()
+        super().__init__()
 
     def _validate_when(self, attr, name, value):
         if not isinstance(value, list):
             setattr(self, name, [value])
 
-    def extract_defined_undefined(self, conditional):
-        results = []
-
-        cond = conditional
-        m = DEFINED_REGEX.search(cond)
-        while m:
-            results.append(m.groups())
-            cond = cond[m.end():]
-            m = DEFINED_REGEX.search(cond)
-
-        return results
-
-    def evaluate_conditional(self, templar, all_vars):
+    def evaluate_conditional(self, templar: Templar, all_vars: dict[str, t.Any]) -> bool:
         '''
         Loops through the conditionals set on this object, returning
         False if any of them evaluate as such.
         '''
-
-        # since this is a mix-in, it may not have an underlying datastructure
-        # associated with it, so we pull it out now in case we need it for
-        # error reporting below
-        ds = None
-        if hasattr(self, '_ds'):
-            ds = getattr(self, '_ds')
-
-        result = True
-        try:
-            for conditional in self.when:
-
-                # do evaluation
-                if conditional is None or conditional == '':
-                    res = True
-                elif isinstance(conditional, bool):
-                    res = conditional
-                else:
+        return self.evaluate_conditional_with_result(templar, all_vars)[0]
+
+    def evaluate_conditional_with_result(self, templar: Templar, all_vars: dict[str, t.Any]) -> tuple[bool, t.Optional[str]]:
+        """Loops through the conditionals set on this object, returning
+        False if any of them evaluate as such as well as the condition
+        that was false.
+        """
+        for conditional in self.when:
+            if conditional is None or conditional == "":
+                res = True
+            elif isinstance(conditional, bool):
+                res = conditional
+            else:
+                try:
                     res = self._check_conditional(conditional, templar, all_vars)
+                except AnsibleError as e:
+                    raise AnsibleError(
+                        "The conditional check '%s' failed. The error was: %s" % (to_native(conditional), to_native(e)),
+                        obj=getattr(self, '_ds', None)
+                    )
 
-                # only update if still true, preserve false
-                if result:
-                    result = res
+            display.debug("Evaluated conditional (%s): %s" % (conditional, res))
+            if not res:
+                return res, conditional
 
-                display.debug("Evaluated conditional (%s): %s" % (conditional, res))
-                if not result:
-                    break
-
-        except Exception as e:
-            raise AnsibleError("The conditional check '%s' failed. The error was: %s" % (to_native(conditional), to_native(e)), obj=ds)
-
-        return result
-
-    def _check_conditional(self, conditional, templar, all_vars):
-        '''
-        This method does the low-level evaluation of each conditional
-        set on this object, using jinja2 to wrap the conditionals for
-        evaluation.
-        '''
+        return True, None
 
+    def _check_conditional(self, conditional: str, templar: Templar, all_vars: dict[str, t.Any]) -> bool:
         original = conditional
-
-        if templar.is_template(conditional):
-            display.warning('conditional statements should not include jinja2 '
-                            'templating delimiters such as {{ }} or {%% %%}. '
-                            'Found: %s' % conditional)
-
-        # make sure the templar is using the variables specified with this method
         templar.available_variables = all_vars
-
         try:
-            # if the conditional is "unsafe", disable lookups
-            disable_lookups = hasattr(conditional, '__UNSAFE__')
-            conditional = templar.template(conditional, disable_lookups=disable_lookups)
-
-            if not isinstance(conditional, text_type) or conditional == "":
-                return conditional
+            if templar.is_template(conditional):
+                display.warning(
+                    "conditional statements should not include jinja2 "
+                    "templating delimiters such as {{ }} or {%% %%}. "
+                    "Found: %s" % conditional
+                )
+                conditional = templar.template(conditional)
+                if isinstance(conditional, bool):
+                    return conditional
+                elif conditional == "":
+                    return False
 
             # If the result of the first-pass template render (to resolve inline templates) is marked unsafe,
             # explicitly fail since the next templating operation would never evaluate
             if hasattr(conditional, '__UNSAFE__'):
                 raise AnsibleTemplateError('Conditional is marked as unsafe, and cannot be evaluated.')
 
-            # First, we do some low-level jinja2 parsing involving the AST format of the
-            # statement to ensure we don't do anything unsafe (using the disable_lookup flag above)
-            class CleansingNodeVisitor(ast.NodeVisitor):
-                def generic_visit(self, node, inside_call=False, inside_yield=False):
-                    if isinstance(node, ast.Call):
-                        inside_call = True
-                    elif isinstance(node, ast.Yield):
-                        inside_yield = True
-                    elif isinstance(node, ast.Str):
-                        if disable_lookups:
-                            if inside_call and node.s.startswith("__"):
-                                # calling things with a dunder is generally bad at this point...
-                                raise AnsibleError(
-                                    "Invalid access found in the conditional: '%s'" % conditional
-                                )
-                            elif inside_yield:
-                                # we're inside a yield, so recursively parse and traverse the AST
-                                # of the result to catch forbidden syntax from executing
-                                parsed = ast.parse(node.s, mode='exec')
-                                cnv = CleansingNodeVisitor()
-                                cnv.visit(parsed)
-                    # iterate over all child nodes
-                    for child_node in ast.iter_child_nodes(node):
-                        self.generic_visit(
-                            child_node,
-                            inside_call=inside_call,
-                            inside_yield=inside_yield
-                        )
-            try:
-                res = templar.environment.parse(conditional, None, None)
-                res = generate(res, templar.environment, None, None)
-                parsed = ast.parse(res, mode='exec')
-
-                cnv = CleansingNodeVisitor()
-                cnv.visit(parsed)
-            except Exception as e:
-                raise AnsibleError("Invalid conditional detected: %s" % to_native(e))
-
-            # and finally we generate and template the presented string and look at the resulting string
             # NOTE The spaces around True and False are intentional to short-circuit literal_eval for
             #      jinja2_native=False and avoid its expensive calls.
-            presented = "{%% if %s %%} True {%% else %%} False {%% endif %%}" % conditional
-            val = templar.template(presented, disable_lookups=disable_lookups).strip()
-            if val == "True":
-                return True
-            elif val == "False":
-                return False
-            else:
-                raise AnsibleError("unable to evaluate conditional: %s" % original)
-        except (AnsibleUndefinedVariable, UndefinedError) as e:
-            # the templating failed, meaning most likely a variable was undefined. If we happened
-            # to be looking for an undefined variable, return True, otherwise fail
-            try:
-                # first we extract the variable name from the error message
-                var_name = re.compile(r"'(hostvars\[.+\]|[\w_]+)' is undefined").search(str(e)).groups()[0]
-                # next we extract all defined/undefined tests from the conditional string
-                def_undef = self.extract_defined_undefined(conditional)
-                # then we loop through these, comparing the error variable name against
-                # each def/undef test we found above. If there is a match, we determine
-                # whether the logic/state mean the variable should exist or not and return
-                # the corresponding True/False
-                for (du_var, logic, state) in def_undef:
-                    # when we compare the var names, normalize quotes because something
-                    # like hostvars['foo'] may be tested against hostvars["foo"]
-                    if var_name.replace("'", '"') == du_var.replace("'", '"'):
-                        # the should exist is a xor test between a negation in the logic portion
-                        # against the state (defined or undefined)
-                        should_exist = ('not' in logic) != (state == 'defined')
-                        if should_exist:
-                            return False
-                        else:
-                            return True
-                # as nothing above matched the failed var name, re-raise here to
-                # trigger the AnsibleUndefinedVariable exception again below
-                raise
-            except Exception:
-                raise AnsibleUndefinedVariable("error while evaluating conditional (%s): %s" % (original, e))
+            return templar.template(
+                "{%% if %s %%} True {%% else %%} False {%% endif %%}" % conditional,
+            ).strip() == "True"
+        except AnsibleUndefinedVariable as e:
+            raise AnsibleUndefinedVariable("error while evaluating conditional (%s): %s" % (original, e))
diff --git a/lib/ansible/playbook/delegatable.py b/lib/ansible/playbook/delegatable.py
new file mode 100644
index 0000000..2d9d16e
--- /dev/null
+++ b/lib/ansible/playbook/delegatable.py
@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+# Copyright The Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from ansible.playbook.attribute import FieldAttribute
+
+
+class Delegatable:
+    delegate_to = FieldAttribute(isa='string')
+    delegate_facts = FieldAttribute(isa='bool')
+
+    def _post_validate_delegate_to(self, attr, value, templar):
+        """This method exists just to make it clear that ``Task.post_validate``
+        does not template this value, it is set via ``TaskExecutor._calculate_delegate_to``
+        """
+        return value
diff --git a/lib/ansible/playbook/handler.py b/lib/ansible/playbook/handler.py
index 68970b4..2f28398 100644
--- a/lib/ansible/playbook/handler.py
+++ b/lib/ansible/playbook/handler.py
@@ -53,6 +53,9 @@ class Handler(Task):
     def remove_host(self, host):
         self.notified_hosts = [h for h in self.notified_hosts if h != host]
 
+    def clear_hosts(self):
+        self.notified_hosts = []
+
     def is_host_notified(self, host):
         return host in self.notified_hosts
 
diff --git a/lib/ansible/playbook/helpers.py b/lib/ansible/playbook/helpers.py
index ff5042a..903dcdf 100644
--- a/lib/ansible/playbook/helpers.py
+++ b/lib/ansible/playbook/helpers.py
@@ -21,9 +21,8 @@ __metaclass__ = type
 import os
 
 from ansible import constants as C
-from ansible.errors import AnsibleParserError, AnsibleUndefinedVariable, AnsibleFileNotFound, AnsibleAssertionError
-from ansible.module_utils._text import to_native
-from ansible.module_utils.six import string_types
+from ansible.errors import AnsibleParserError, AnsibleUndefinedVariable, AnsibleAssertionError
+from ansible.module_utils.common.text.converters import to_native
 from ansible.parsing.mod_args import ModuleArgsParser
 from ansible.utils.display import Display
 
@@ -151,23 +150,9 @@ def load_list_of_tasks(ds, play, block=None, role=None, task_include=None, use_h
                 templar = Templar(loader=loader, variables=all_vars)
 
                 # check to see if this include is dynamic or static:
-                # 1. the user has set the 'static' option to false or true
-                # 2. one of the appropriate config options was set
-                if action in C._ACTION_INCLUDE_TASKS:
-                    is_static = False
-                elif action in C._ACTION_IMPORT_TASKS:
-                    is_static = True
-                else:
-                    include_link = get_versioned_doclink('user_guide/playbooks_reuse_includes.html')
-                    display.deprecated('"include" is deprecated, use include_tasks/import_tasks instead. See %s for details' % include_link, "2.16")
-                    is_static = not templar.is_template(t.args['_raw_params']) and t.all_parents_static() and not t.loop
-
-                if is_static:
+                if action in C._ACTION_IMPORT_TASKS:
                     if t.loop is not None:
-                        if action in C._ACTION_IMPORT_TASKS:
-                            raise AnsibleParserError("You cannot use loops on 'import_tasks' statements. You should use 'include_tasks' instead.", obj=task_ds)
-                        else:
-                            raise AnsibleParserError("You cannot use 'static' on an include with a loop", obj=task_ds)
+                        raise AnsibleParserError("You cannot use loops on 'import_tasks' statements. You should use 'include_tasks' instead.", obj=task_ds)
 
                     # we set a flag to indicate this include was static
                     t.statically_loaded = True
@@ -289,18 +274,9 @@ def load_list_of_tasks(ds, play, block=None, role=None, task_include=None, use_h
                     loader=loader,
                 )
 
-                #   1. the user has set the 'static' option to false or true
-                #   2. one of the appropriate config options was set
-                is_static = False
                 if action in C._ACTION_IMPORT_ROLE:
-                    is_static = True
-
-                if is_static:
                     if ir.loop is not None:
-                        if action in C._ACTION_IMPORT_ROLE:
-                            raise AnsibleParserError("You cannot use loops on 'import_role' statements. You should use 'include_role' instead.", obj=task_ds)
-                        else:
-                            raise AnsibleParserError("You cannot use 'static' on an include_role with a loop", obj=task_ds)
+                        raise AnsibleParserError("You cannot use loops on 'import_role' statements. You should use 'include_role' instead.", obj=task_ds)
 
                     # we set a flag to indicate this include was static
                     ir.statically_loaded = True
@@ -312,7 +288,7 @@ def load_list_of_tasks(ds, play, block=None, role=None, task_include=None, use_h
                     ir._role_name = templar.template(ir._role_name)
 
                     # uses compiled list from object
-                    blocks, _ = ir.get_block_list(variable_manager=variable_manager, loader=loader)
+                    blocks, dummy = ir.get_block_list(variable_manager=variable_manager, loader=loader)
                     task_list.extend(blocks)
                 else:
                     # passes task object itself for latter generation of list
diff --git a/lib/ansible/playbook/included_file.py b/lib/ansible/playbook/included_file.py
index b833077..925d439 100644
--- a/lib/ansible/playbook/included_file.py
+++ b/lib/ansible/playbook/included_file.py
@@ -24,7 +24,7 @@ import os
 from ansible import constants as C
 from ansible.errors import AnsibleError
 from ansible.executor.task_executor import remove_omit
-from ansible.module_utils._text import to_text
+from ansible.module_utils.common.text.converters import to_text
 from ansible.playbook.handler import Handler
 from ansible.playbook.task_include import TaskInclude
 from ansible.playbook.role_include import IncludeRole
@@ -72,8 +72,6 @@ class IncludedFile:
             original_task = res._task
 
             if original_task.action in C._ACTION_ALL_INCLUDES:
-                if original_task.action in C._ACTION_INCLUDE:
-                    display.deprecated('"include" is deprecated, use include_tasks/import_tasks/import_playbook instead', "2.16")
 
                 if original_task.loop:
                     if 'results' not in res._result:
@@ -118,7 +116,7 @@ class IncludedFile:
 
                     templar = Templar(loader=loader, variables=task_vars)
 
-                    if original_task.action in C._ACTION_ALL_INCLUDE_TASKS:
+                    if original_task.action in C._ACTION_INCLUDE_TASKS:
                         include_file = None
 
                         if original_task._parent:
@@ -148,9 +146,12 @@ class IncludedFile:
                                     cumulative_path = parent_include_dir
                                 include_target = templar.template(include_result['include'])
                                 if original_task._role:
-                                    new_basedir = os.path.join(original_task._role._role_path, 'tasks', cumulative_path)
-                                    candidates = [loader.path_dwim_relative(original_task._role._role_path, 'tasks', include_target),
-                                                  loader.path_dwim_relative(new_basedir, 'tasks', include_target)]
+                                    dirname = 'handlers' if isinstance(original_task, Handler) else 'tasks'
+                                    new_basedir = os.path.join(original_task._role._role_path, dirname, cumulative_path)
+                                    candidates = [
+                                        loader.path_dwim_relative(original_task._role._role_path, dirname, include_target, is_role=True),
+                                        loader.path_dwim_relative(new_basedir, dirname, include_target, is_role=True)
+                                    ]
                                     for include_file in candidates:
                                         try:
                                             # may throw OSError
diff --git a/lib/ansible/playbook/loop_control.py b/lib/ansible/playbook/loop_control.py
index d69e14f..4df0a73 100644
--- a/lib/ansible/playbook/loop_control.py
+++ b/lib/ansible/playbook/loop_control.py
@@ -25,9 +25,9 @@ from ansible.playbook.base import FieldAttributeBase
 
 class LoopControl(FieldAttributeBase):
 
-    loop_var = NonInheritableFieldAttribute(isa='str', default='item', always_post_validate=True)
-    index_var = NonInheritableFieldAttribute(isa='str', always_post_validate=True)
-    label = NonInheritableFieldAttribute(isa='str')
+    loop_var = NonInheritableFieldAttribute(isa='string', default='item', always_post_validate=True)
+    index_var = NonInheritableFieldAttribute(isa='string', always_post_validate=True)
+    label = NonInheritableFieldAttribute(isa='string')
     pause = NonInheritableFieldAttribute(isa='float', default=0, always_post_validate=True)
     extended = NonInheritableFieldAttribute(isa='bool', always_post_validate=True)
     extended_allitems = NonInheritableFieldAttribute(isa='bool', default=True, always_post_validate=True)
diff --git a/lib/ansible/playbook/notifiable.py b/lib/ansible/playbook/notifiable.py
new file mode 100644
index 0000000..a183293
--- /dev/null
+++ b/lib/ansible/playbook/notifiable.py
@@ -0,0 +1,9 @@
+# -*- coding: utf-8 -*-
+# Copyright The Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from ansible.playbook.attribute import FieldAttribute
+
+
+class Notifiable:
+    notify = FieldAttribute(isa='list')
diff --git a/lib/ansible/playbook/play.py b/lib/ansible/playbook/play.py
index 3b763b9..6449859 100644
--- a/lib/ansible/playbook/play.py
+++ b/lib/ansible/playbook/play.py
@@ -22,7 +22,7 @@ __metaclass__ = type
 from ansible import constants as C
 from ansible import context
 from ansible.errors import AnsibleParserError, AnsibleAssertionError
-from ansible.module_utils._text import to_native
+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
 from ansible.playbook.attribute import NonInheritableFieldAttribute
@@ -30,7 +30,7 @@ from ansible.playbook.base import Base
 from ansible.playbook.block import Block
 from ansible.playbook.collectionsearch import CollectionSearch
 from ansible.playbook.helpers import load_list_of_blocks, load_list_of_roles
-from ansible.playbook.role import Role
+from ansible.playbook.role import Role, hash_params
 from ansible.playbook.task import Task
 from ansible.playbook.taggable import Taggable
 from ansible.vars.manager import preprocess_vars
@@ -93,7 +93,7 @@ class Play(Base, Taggable, CollectionSearch):
         self._included_conditional = None
         self._included_path = None
         self._removed_hosts = []
-        self.ROLE_CACHE = {}
+        self.role_cache = {}
 
         self.only_tags = set(context.CLIARGS.get('tags', [])) or frozenset(('all',))
         self.skip_tags = set(context.CLIARGS.get('skip_tags', []))
@@ -104,6 +104,22 @@ class Play(Base, Taggable, CollectionSearch):
     def __repr__(self):
         return self.get_name()
 
+    @property
+    def ROLE_CACHE(self):
+        """Backwards compat for custom strategies using ``play.ROLE_CACHE``
+        """
+        display.deprecated(
+            'Play.ROLE_CACHE is deprecated in favor of Play.role_cache, or StrategyBase._get_cached_role',
+            version='2.18',
+        )
+        cache = {}
+        for path, roles in self.role_cache.items():
+            for role in roles:
+                name = role.get_name()
+                hashed_params = hash_params(role._get_hash_dict())
+                cache.setdefault(name, {})[hashed_params] = role
+        return 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:
@@ -393,7 +409,7 @@ class Play(Base, Taggable, CollectionSearch):
 
     def copy(self):
         new_me = super(Play, self).copy()
-        new_me.ROLE_CACHE = self.ROLE_CACHE.copy()
+        new_me.role_cache = self.role_cache.copy()
         new_me._included_conditional = self._included_conditional
         new_me._included_path = self._included_path
         new_me._action_groups = self._action_groups
diff --git a/lib/ansible/playbook/play_context.py b/lib/ansible/playbook/play_context.py
index 90de929..af65e86 100644
--- a/lib/ansible/playbook/play_context.py
+++ b/lib/ansible/playbook/play_context.py
@@ -23,11 +23,9 @@ __metaclass__ = type
 
 from ansible import constants as C
 from ansible import context
-from ansible.module_utils.compat.paramiko import paramiko
 from ansible.playbook.attribute import FieldAttribute
 from ansible.playbook.base import Base
 from ansible.utils.display import Display
-from ansible.utils.ssh_functions import check_for_controlpersist
 
 
 display = Display()
@@ -121,7 +119,7 @@ class PlayContext(Base):
     def verbosity(self):
         display.deprecated(
             "PlayContext.verbosity is deprecated, use ansible.utils.display.Display.verbosity instead.",
-            version=2.18
+            version="2.18"
         )
         return self._internal_verbosity
 
@@ -129,7 +127,7 @@ class PlayContext(Base):
     def verbosity(self, value):
         display.deprecated(
             "PlayContext.verbosity is deprecated, use ansible.utils.display.Display.verbosity instead.",
-            version=2.18
+            version="2.18"
         )
         self._internal_verbosity = value
 
@@ -320,10 +318,6 @@ class PlayContext(Base):
             display.warning('The "%s" connection plugin has an improperly configured remote target value, '
                             'forcing "inventory_hostname" templated value instead of the string' % new_info.connection)
 
-        # set no_log to default if it was not previously set
-        if new_info.no_log is None:
-            new_info.no_log = C.DEFAULT_NO_LOG
-
         if task.check_mode is not None:
             new_info.check_mode = task.check_mode
 
diff --git a/lib/ansible/playbook/playbook_include.py b/lib/ansible/playbook/playbook_include.py
index 8e3116f..2579a8a 100644
--- a/lib/ansible/playbook/playbook_include.py
+++ b/lib/ansible/playbook/playbook_include.py
@@ -23,9 +23,9 @@ import os
 
 import ansible.constants as C
 from ansible.errors import AnsibleParserError, AnsibleAssertionError
-from ansible.module_utils._text import to_bytes
+from ansible.module_utils.common.text.converters import to_bytes
 from ansible.module_utils.six import string_types
-from ansible.parsing.splitter import split_args, parse_kv
+from ansible.parsing.splitter import split_args
 from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject, AnsibleMapping
 from ansible.playbook.attribute import NonInheritableFieldAttribute
 from ansible.playbook.base import Base
@@ -48,7 +48,7 @@ class PlaybookInclude(Base, Conditional, Taggable):
     def load(data, basedir, variable_manager=None, loader=None):
         return PlaybookInclude().load_data(ds=data, basedir=basedir, variable_manager=variable_manager, loader=loader)
 
-    def load_data(self, ds, basedir, variable_manager=None, loader=None):
+    def load_data(self, ds, variable_manager=None, loader=None, basedir=None):
         '''
         Overrides the base load_data(), as we're actually going to return a new
         Playbook() object rather than a PlaybookInclude object
diff --git a/lib/ansible/playbook/role/__init__.py b/lib/ansible/playbook/role/__init__.py
index 0409609..34d8ba9 100644
--- a/lib/ansible/playbook/role/__init__.py
+++ b/lib/ansible/playbook/role/__init__.py
@@ -22,15 +22,17 @@ __metaclass__ = type
 import os
 
 from collections.abc import Container, Mapping, Set, Sequence
+from types import MappingProxyType
 
 from ansible import constants as C
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleAssertionError
-from ansible.module_utils._text import to_text
+from ansible.module_utils.common.text.converters import to_text
 from ansible.module_utils.six import binary_type, text_type
 from ansible.playbook.attribute import FieldAttribute
 from ansible.playbook.base import Base
 from ansible.playbook.collectionsearch import CollectionSearch
 from ansible.playbook.conditional import Conditional
+from ansible.playbook.delegatable import Delegatable
 from ansible.playbook.helpers import load_list_of_blocks
 from ansible.playbook.role.metadata import RoleMetadata
 from ansible.playbook.taggable import Taggable
@@ -96,22 +98,32 @@ def hash_params(params):
     return frozenset((params,))
 
 
-class Role(Base, Conditional, Taggable, CollectionSearch):
+class Role(Base, Conditional, Taggable, CollectionSearch, Delegatable):
 
-    delegate_to = FieldAttribute(isa='string')
-    delegate_facts = FieldAttribute(isa='bool')
-
-    def __init__(self, play=None, from_files=None, from_include=False, validate=True):
+    def __init__(self, play=None, from_files=None, from_include=False, validate=True, public=None, static=True):
         self._role_name = None
         self._role_path = None
         self._role_collection = None
         self._role_params = dict()
         self._loader = None
+        self.static = static
+
+        # includes (static=false) default to private, while imports (static=true) default to public
+        # but both can be overriden by global config if set
+        if public is None:
+            global_private, origin = C.config.get_config_value_and_origin('DEFAULT_PRIVATE_ROLE_VARS')
+            if origin == 'default':
+                self.public = static
+            else:
+                self.public = not global_private
+        else:
+            self.public = public
 
-        self._metadata = None
+        self._metadata = RoleMetadata()
         self._play = play
         self._parents = []
         self._dependencies = []
+        self._all_dependencies = None
         self._task_blocks = []
         self._handler_blocks = []
         self._compiled_handler_blocks = None
@@ -128,6 +140,8 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         # Indicates whether this role was included via include/import_role
         self.from_include = from_include
 
+        self._hash = None
+
         super(Role, self).__init__()
 
     def __repr__(self):
@@ -138,49 +152,54 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
             return '.'.join(x for x in (self._role_collection, self._role_name) if x)
         return self._role_name
 
-    @staticmethod
-    def load(role_include, play, parent_role=None, from_files=None, from_include=False, validate=True):
+    def get_role_path(self):
+        # Purposefully using realpath for canonical path
+        return os.path.realpath(self._role_path)
+
+    def _get_hash_dict(self):
+        if self._hash:
+            return self._hash
+        self._hash = MappingProxyType(
+            {
+                'name': self.get_name(),
+                'path': self.get_role_path(),
+                'params': MappingProxyType(self.get_role_params()),
+                'when': self.when,
+                'tags': self.tags,
+                'from_files': MappingProxyType(self._from_files),
+                'vars': MappingProxyType(self.vars),
+                'from_include': self.from_include,
+            }
+        )
+        return self._hash
+
+    def __eq__(self, other):
+        if not isinstance(other, Role):
+            return False
+
+        return self._get_hash_dict() == other._get_hash_dict()
 
+    @staticmethod
+    def load(role_include, play, parent_role=None, from_files=None, from_include=False, validate=True, public=None, static=True):
         if from_files is None:
             from_files = {}
         try:
-            # The ROLE_CACHE is a dictionary of role names, with each entry
-            # containing another dictionary corresponding to a set of parameters
-            # specified for a role as the key and the Role() object itself.
-            # We use frozenset to make the dictionary hashable.
-
-            params = role_include.get_role_params()
-            if role_include.when is not None:
-                params['when'] = role_include.when
-            if role_include.tags is not None:
-                params['tags'] = role_include.tags
-            if from_files is not None:
-                params['from_files'] = from_files
-            if role_include.vars:
-                params['vars'] = role_include.vars
-
-            params['from_include'] = from_include
-
-            hashed_params = hash_params(params)
-            if role_include.get_name() in play.ROLE_CACHE:
-                for (entry, role_obj) in play.ROLE_CACHE[role_include.get_name()].items():
-                    if hashed_params == entry:
-                        if parent_role:
-                            role_obj.add_parent(parent_role)
-                        return role_obj
-
             # TODO: need to fix cycle detection in role load (maybe use an empty dict
             #  for the in-flight in role cache as a sentinel that we're already trying to load
             #  that role?)
             # see https://github.com/ansible/ansible/issues/61527
-            r = Role(play=play, from_files=from_files, from_include=from_include, validate=validate)
+            r = Role(play=play, from_files=from_files, from_include=from_include, validate=validate, public=public, static=static)
             r._load_role_data(role_include, parent_role=parent_role)
 
-            if role_include.get_name() not in play.ROLE_CACHE:
-                play.ROLE_CACHE[role_include.get_name()] = dict()
+            role_path = r.get_role_path()
+            if role_path not in play.role_cache:
+                play.role_cache[role_path] = []
+
+            # Using the role path as a cache key is done to improve performance when a large number of roles
+            # are in use in the play
+            if r not in play.role_cache[role_path]:
+                play.role_cache[role_path].append(r)
 
-            # FIXME: how to handle cache keys for collection-based roles, since they're technically adjustable per task?
-            play.ROLE_CACHE[role_include.get_name()][hashed_params] = r
             return r
 
         except RuntimeError:
@@ -221,8 +240,6 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         if metadata:
             self._metadata = RoleMetadata.load(metadata, owner=self, variable_manager=self._variable_manager, loader=self._loader)
             self._dependencies = self._load_dependencies()
-        else:
-            self._metadata = RoleMetadata()
 
         # reset collections list; roles do not inherit collections from parents, just use the defaults
         # FUTURE: use a private config default for this so we can allow it to be overridden later
@@ -421,10 +438,9 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         '''
 
         deps = []
-        if self._metadata:
-            for role_include in self._metadata.dependencies:
-                r = Role.load(role_include, play=self._play, parent_role=self)
-                deps.append(r)
+        for role_include in self._metadata.dependencies:
+            r = Role.load(role_include, play=self._play, parent_role=self, static=self.static)
+            deps.append(r)
 
         return deps
 
@@ -441,6 +457,13 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
     def get_parents(self):
         return self._parents
 
+    def get_dep_chain(self):
+        dep_chain = []
+        for parent in self._parents:
+            dep_chain.extend(parent.get_dep_chain())
+            dep_chain.append(parent)
+        return dep_chain
+
     def get_default_vars(self, dep_chain=None):
         dep_chain = [] if dep_chain is None else dep_chain
 
@@ -453,14 +476,15 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         default_vars = combine_vars(default_vars, self._default_vars)
         return default_vars
 
-    def get_inherited_vars(self, dep_chain=None):
+    def get_inherited_vars(self, dep_chain=None, only_exports=False):
         dep_chain = [] if dep_chain is None else dep_chain
 
         inherited_vars = dict()
 
         if dep_chain:
             for parent in dep_chain:
-                inherited_vars = combine_vars(inherited_vars, parent.vars)
+                if not only_exports:
+                    inherited_vars = combine_vars(inherited_vars, parent.vars)
                 inherited_vars = combine_vars(inherited_vars, parent._role_vars)
         return inherited_vars
 
@@ -474,18 +498,36 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         params = combine_vars(params, self._role_params)
         return params
 
-    def get_vars(self, dep_chain=None, include_params=True):
+    def get_vars(self, dep_chain=None, include_params=True, only_exports=False):
         dep_chain = [] if dep_chain is None else dep_chain
 
-        all_vars = self.get_inherited_vars(dep_chain)
+        all_vars = {}
 
-        for dep in self.get_all_dependencies():
-            all_vars = combine_vars(all_vars, dep.get_vars(include_params=include_params))
+        # get role_vars: from parent objects
+        # TODO: is this right precedence for inherited role_vars?
+        all_vars = self.get_inherited_vars(dep_chain, only_exports=only_exports)
 
-        all_vars = combine_vars(all_vars, self.vars)
+        # get exported variables from meta/dependencies
+        seen = []
+        for dep in self.get_all_dependencies():
+            # Avoid reruning dupe deps since they can have vars from previous invocations and they accumulate in deps
+            # TODO: re-examine dep loading to see if we are somehow improperly adding the same dep too many times
+            if dep not in seen:
+                # only take 'exportable' vars from deps
+                all_vars = combine_vars(all_vars, dep.get_vars(include_params=False, only_exports=True))
+                seen.append(dep)
+
+        # role_vars come from vars/ in a role
         all_vars = combine_vars(all_vars, self._role_vars)
-        if include_params:
-            all_vars = combine_vars(all_vars, self.get_role_params(dep_chain=dep_chain))
+
+        if not only_exports:
+            # include_params are 'inline variables' in role invocation. - {role: x, varname: value}
+            if include_params:
+                # TODO: add deprecation notice
+                all_vars = combine_vars(all_vars, self.get_role_params(dep_chain=dep_chain))
+
+            # these come from vars: keyword in role invocation. - {role: x, vars: {varname: value}}
+            all_vars = combine_vars(all_vars, self.vars)
 
         return all_vars
 
@@ -497,15 +539,15 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         Returns a list of all deps, built recursively from all child dependencies,
         in the proper order in which they should be executed or evaluated.
         '''
+        if self._all_dependencies is None:
 
-        child_deps = []
-
-        for dep in self.get_direct_dependencies():
-            for child_dep in dep.get_all_dependencies():
-                child_deps.append(child_dep)
-            child_deps.append(dep)
+            self._all_dependencies = []
+            for dep in self.get_direct_dependencies():
+                for child_dep in dep.get_all_dependencies():
+                    self._all_dependencies.append(child_dep)
+                self._all_dependencies.append(dep)
 
-        return child_deps
+        return self._all_dependencies
 
     def get_task_blocks(self):
         return self._task_blocks[:]
@@ -607,8 +649,7 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
         res['_had_task_run'] = self._had_task_run.copy()
         res['_completed'] = self._completed.copy()
 
-        if self._metadata:
-            res['_metadata'] = self._metadata.serialize()
+        res['_metadata'] = self._metadata.serialize()
 
         if include_deps:
             deps = []
diff --git a/lib/ansible/playbook/role/include.py b/lib/ansible/playbook/role/include.py
index e0d4b67..f4b3e40 100644
--- a/lib/ansible/playbook/role/include.py
+++ b/lib/ansible/playbook/role/include.py
@@ -22,24 +22,21 @@ __metaclass__ = type
 from ansible.errors import AnsibleError, AnsibleParserError
 from ansible.module_utils.six import string_types
 from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject
-from ansible.playbook.attribute import FieldAttribute
+from ansible.playbook.delegatable import Delegatable
 from ansible.playbook.role.definition import RoleDefinition
-from ansible.module_utils._text import to_native
+from ansible.module_utils.common.text.converters import to_native
 
 
 __all__ = ['RoleInclude']
 
 
-class RoleInclude(RoleDefinition):
+class RoleInclude(RoleDefinition, Delegatable):
 
     """
     A derivative of RoleDefinition, used by playbook code when a role
     is included for execution in a play.
     """
 
-    delegate_to = FieldAttribute(isa='string')
-    delegate_facts = FieldAttribute(isa='bool', default=False)
-
     def __init__(self, play=None, role_basedir=None, variable_manager=None, loader=None, collection_list=None):
         super(RoleInclude, self).__init__(play=play, role_basedir=role_basedir, variable_manager=variable_manager,
                                           loader=loader, collection_list=collection_list)
diff --git a/lib/ansible/playbook/role/metadata.py b/lib/ansible/playbook/role/metadata.py
index a4dbcf7..e299122 100644
--- a/lib/ansible/playbook/role/metadata.py
+++ b/lib/ansible/playbook/role/metadata.py
@@ -22,7 +22,7 @@ __metaclass__ = type
 import os
 
 from ansible.errors import AnsibleParserError, AnsibleError
-from ansible.module_utils._text import to_native
+from ansible.module_utils.common.text.converters import to_native
 from ansible.module_utils.six import string_types
 from ansible.playbook.attribute import NonInheritableFieldAttribute
 from ansible.playbook.base import Base
@@ -41,7 +41,7 @@ class RoleMetadata(Base, CollectionSearch):
 
     allow_duplicates = NonInheritableFieldAttribute(isa='bool', default=False)
     dependencies = NonInheritableFieldAttribute(isa='list', default=list)
-    galaxy_info = NonInheritableFieldAttribute(isa='GalaxyInfo')
+    galaxy_info = NonInheritableFieldAttribute(isa='dict')
     argument_specs = NonInheritableFieldAttribute(isa='dict', default=dict)
 
     def __init__(self, owner=None):
@@ -110,15 +110,6 @@ class RoleMetadata(Base, CollectionSearch):
         except AssertionError as e:
             raise AnsibleParserError("A malformed list of role dependencies was encountered.", obj=self._ds, orig_exc=e)
 
-    def _load_galaxy_info(self, attr, ds):
-        '''
-        This is a helper loading function for the galaxy info entry
-        in the metadata, which returns a GalaxyInfo object rather than
-        a simple dictionary.
-        '''
-
-        return ds
-
     def serialize(self):
         return dict(
             allow_duplicates=self._allow_duplicates,
diff --git a/lib/ansible/playbook/role_include.py b/lib/ansible/playbook/role_include.py
index 75d26fb..cdf86c0 100644
--- a/lib/ansible/playbook/role_include.py
+++ b/lib/ansible/playbook/role_include.py
@@ -23,7 +23,6 @@ from os.path import basename
 import ansible.constants as C
 from ansible.errors import AnsibleParserError
 from ansible.playbook.attribute import NonInheritableFieldAttribute
-from ansible.playbook.block import Block
 from ansible.playbook.task_include import TaskInclude
 from ansible.playbook.role import Role
 from ansible.playbook.role.include import RoleInclude
@@ -50,10 +49,10 @@ class IncludeRole(TaskInclude):
 
     # =================================================================================
     # ATTRIBUTES
+    public = NonInheritableFieldAttribute(isa='bool', default=None, private=False, always_post_validate=True)
 
     # private as this is a 'module options' vs a task property
     allow_duplicates = NonInheritableFieldAttribute(isa='bool', default=True, private=True, always_post_validate=True)
-    public = NonInheritableFieldAttribute(isa='bool', default=False, private=True, always_post_validate=True)
     rolespec_validate = NonInheritableFieldAttribute(isa='bool', default=True, private=True, always_post_validate=True)
 
     def __init__(self, block=None, role=None, task_include=None):
@@ -89,22 +88,18 @@ class IncludeRole(TaskInclude):
 
         # build role
         actual_role = Role.load(ri, myplay, parent_role=self._parent_role, from_files=from_files,
-                                from_include=True, validate=self.rolespec_validate)
+                                from_include=True, validate=self.rolespec_validate, public=self.public, static=self.statically_loaded)
         actual_role._metadata.allow_duplicates = self.allow_duplicates
 
-        if self.statically_loaded or self.public:
-            myplay.roles.append(actual_role)
+        # add role to play
+        myplay.roles.append(actual_role)
 
         # save this for later use
         self._role_path = actual_role._role_path
 
         # compile role with parent roles as dependencies to ensure they inherit
         # variables
-        if not self._parent_role:
-            dep_chain = []
-        else:
-            dep_chain = list(self._parent_role._parents)
-            dep_chain.append(self._parent_role)
+        dep_chain = actual_role.get_dep_chain()
 
         p_block = self.build_parent_block()
 
@@ -118,7 +113,7 @@ class IncludeRole(TaskInclude):
             b.collections = actual_role.collections
 
         # updated available handlers in play
-        handlers = actual_role.get_handler_blocks(play=myplay)
+        handlers = actual_role.get_handler_blocks(play=myplay, dep_chain=dep_chain)
         for h in handlers:
             h._parent = p_block
         myplay.handlers = myplay.handlers + handlers
@@ -137,6 +132,7 @@ class IncludeRole(TaskInclude):
         if ir._role_name is None:
             raise AnsibleParserError("'name' is a required field for %s." % ir.action, obj=data)
 
+        # public is only valid argument for includes, imports are always 'public' (after they run)
         if 'public' in ir.args and ir.action not in C._ACTION_INCLUDE_ROLE:
             raise AnsibleParserError('Invalid options for %s: public' % ir.action, obj=data)
 
@@ -145,7 +141,7 @@ class IncludeRole(TaskInclude):
         if bad_opts:
             raise AnsibleParserError('Invalid options for %s: %s' % (ir.action, ','.join(list(bad_opts))), obj=data)
 
-        # build options for role includes
+        # build options for role include/import tasks
         for key in my_arg_names.intersection(IncludeRole.FROM_ARGS):
             from_key = key.removesuffix('_from')
             args_value = ir.args.get(key)
@@ -153,6 +149,7 @@ class IncludeRole(TaskInclude):
                 raise AnsibleParserError('Expected a string for %s but got %s instead' % (key, type(args_value)))
             ir._from_files[from_key] = basename(args_value)
 
+        # apply is only valid for includes, not imports as they inherit directly
         apply_attrs = ir.args.get('apply', {})
         if apply_attrs and ir.action not in C._ACTION_INCLUDE_ROLE:
             raise AnsibleParserError('Invalid options for %s: apply' % ir.action, obj=data)
diff --git a/lib/ansible/playbook/taggable.py b/lib/ansible/playbook/taggable.py
index 4038d7f..828c7b2 100644
--- a/lib/ansible/playbook/taggable.py
+++ b/lib/ansible/playbook/taggable.py
@@ -23,6 +23,17 @@ from ansible.errors import AnsibleError
 from ansible.module_utils.six import string_types
 from ansible.playbook.attribute import FieldAttribute
 from ansible.template import Templar
+from ansible.utils.sentinel import Sentinel
+
+
+def _flatten_tags(tags: list) -> list:
+    rv = set()
+    for tag in tags:
+        if isinstance(tag, list):
+            rv.update(tag)
+        else:
+            rv.add(tag)
+    return list(rv)
 
 
 class Taggable:
@@ -34,11 +45,7 @@ class Taggable:
         if isinstance(ds, list):
             return ds
         elif isinstance(ds, string_types):
-            value = ds.split(',')
-            if isinstance(value, list):
-                return [x.strip() for x in value]
-            else:
-                return [ds]
+            return [x.strip() for x in ds.split(',')]
         else:
             raise AnsibleError('tags must be specified as a list', obj=ds)
 
@@ -47,16 +54,12 @@ class Taggable:
 
         if self.tags:
             templar = Templar(loader=self._loader, variables=all_vars)
-            tags = templar.template(self.tags)
-
-            _temp_tags = set()
-            for tag in tags:
-                if isinstance(tag, list):
-                    _temp_tags.update(tag)
-                else:
-                    _temp_tags.add(tag)
-            tags = _temp_tags
-            self.tags = list(tags)
+            obj = self
+            while obj is not None:
+                if (_tags := getattr(obj, "_tags", Sentinel)) is not Sentinel:
+                    obj._tags = _flatten_tags(templar.template(_tags))
+                obj = obj._parent
+            tags = set(self.tags)
         else:
             # this makes isdisjoint work for untagged
             tags = self.untagged
diff --git a/lib/ansible/playbook/task.py b/lib/ansible/playbook/task.py
index a1a1162..fa1114a 100644
--- a/lib/ansible/playbook/task.py
+++ b/lib/ansible/playbook/task.py
@@ -21,17 +21,19 @@ __metaclass__ = type
 
 from ansible import constants as C
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleAssertionError
-from ansible.module_utils._text import to_native
+from ansible.module_utils.common.text.converters import to_native
 from ansible.module_utils.six import string_types
 from ansible.parsing.mod_args import ModuleArgsParser
 from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject, AnsibleMapping
 from ansible.plugins.loader import lookup_loader
-from ansible.playbook.attribute import FieldAttribute, NonInheritableFieldAttribute
+from ansible.playbook.attribute import NonInheritableFieldAttribute
 from ansible.playbook.base import Base
 from ansible.playbook.block import Block
 from ansible.playbook.collectionsearch import CollectionSearch
 from ansible.playbook.conditional import Conditional
+from ansible.playbook.delegatable import Delegatable
 from ansible.playbook.loop_control import LoopControl
+from ansible.playbook.notifiable import Notifiable
 from ansible.playbook.role import Role
 from ansible.playbook.taggable import Taggable
 from ansible.utils.collection_loader import AnsibleCollectionConfig
@@ -43,7 +45,7 @@ __all__ = ['Task']
 display = Display()
 
 
-class Task(Base, Conditional, Taggable, CollectionSearch):
+class Task(Base, Conditional, Taggable, CollectionSearch, Notifiable, Delegatable):
 
     """
     A task is a language feature that represents a call to a module, with given arguments and other parameters.
@@ -72,15 +74,12 @@ class Task(Base, Conditional, Taggable, CollectionSearch):
     async_val = NonInheritableFieldAttribute(isa='int', default=0, alias='async')
     changed_when = NonInheritableFieldAttribute(isa='list', default=list)
     delay = NonInheritableFieldAttribute(isa='int', default=5)
-    delegate_to = FieldAttribute(isa='string')
-    delegate_facts = FieldAttribute(isa='bool')
     failed_when = NonInheritableFieldAttribute(isa='list', default=list)
-    loop = NonInheritableFieldAttribute()
+    loop = NonInheritableFieldAttribute(isa='list')
     loop_control = NonInheritableFieldAttribute(isa='class', class_type=LoopControl, default=LoopControl)
-    notify = FieldAttribute(isa='list')
     poll = NonInheritableFieldAttribute(isa='int', default=C.DEFAULT_POLL_INTERVAL)
     register = NonInheritableFieldAttribute(isa='string', static=True)
-    retries = NonInheritableFieldAttribute(isa='int', default=3)
+    retries = NonInheritableFieldAttribute(isa='int')  # default is set in TaskExecutor
     until = NonInheritableFieldAttribute(isa='list', default=list)
 
     # deprecated, used to be loop and loop_args but loop has been repurposed
@@ -138,7 +137,7 @@ class Task(Base, Conditional, Taggable, CollectionSearch):
 
     def __repr__(self):
         ''' returns a human readable representation of the task '''
-        if self.get_name() in C._ACTION_META:
+        if self.action in C._ACTION_META:
             return "TASK: meta (%s)" % self.args['_raw_params']
         else:
             return "TASK: %s" % self.get_name()
@@ -533,3 +532,9 @@ class Task(Base, Conditional, Taggable, CollectionSearch):
                 return self._parent
             return self._parent.get_first_parent_include()
         return None
+
+    def get_play(self):
+        parent = self._parent
+        while not isinstance(parent, Block):
+            parent = parent._parent
+        return parent._play
diff --git a/lib/ansible/playbook/task_include.py b/lib/ansible/playbook/task_include.py
index 9c335c6..fc09889 100644
--- a/lib/ansible/playbook/task_include.py
+++ b/lib/ansible/playbook/task_include.py
@@ -35,7 +35,7 @@ class TaskInclude(Task):
 
     """
     A task include is derived from a regular task to handle the special
-    circumstances related to the `- include: ...` task.
+    circumstances related to the `- include_*: ...` task.
     """
 
     BASE = frozenset(('file', '_raw_params'))  # directly assigned
@@ -105,29 +105,6 @@ class TaskInclude(Task):
         new_me.statically_loaded = self.statically_loaded
         return new_me
 
-    def get_vars(self):
-        '''
-        We override the parent Task() classes get_vars here because
-        we need to include the args of the include into the vars as
-        they are params to the included tasks. But ONLY for 'include'
-        '''
-        if self.action not in C._ACTION_INCLUDE:
-            all_vars = super(TaskInclude, self).get_vars()
-        else:
-            all_vars = dict()
-            if self._parent:
-                all_vars |= self._parent.get_vars()
-
-            all_vars |= self.vars
-            all_vars |= self.args
-
-            if 'tags' in all_vars:
-                del all_vars['tags']
-            if 'when' in all_vars:
-                del all_vars['when']
-
-        return all_vars
-
     def build_parent_block(self):
         '''
         This method is used to create the parent block for the included tasks