Adding upstream version 2.14.3.upstream/2.14.3upstream

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

25 files changed, 6559 insertions, 0 deletions

diff --git a/lib/ansible/executor/__init__.py b/lib/ansible/executor/__init__.py
new file mode 100644
index 0000000..ae8ccff
--- /dev/null
+++ b/lib/ansible/executor/__init__.py
@@ -0,0 +1,20 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
diff --git a/lib/ansible/executor/action_write_locks.py b/lib/ansible/executor/action_write_locks.py
new file mode 100644
index 0000000..fd82744
--- /dev/null
+++ b/lib/ansible/executor/action_write_locks.py
@@ -0,0 +1,46 @@
+# (c) 2016 - Red Hat, Inc. <info@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import multiprocessing.synchronize
+
+from multiprocessing import Lock
+
+from ansible.module_utils.facts.system.pkg_mgr import PKG_MGRS
+
+if 'action_write_locks' not in globals():
+    # Do not initialize this more than once because it seems to bash
+    # the existing one.  multiprocessing must be reloading the module
+    # when it forks?
+    action_write_locks = dict()  # type: dict[str | None, multiprocessing.synchronize.Lock]
+
+    # Below is a Lock for use when we weren't expecting a named module.  It gets used when an action
+    # plugin invokes a module whose name does not match with the action's name.  Slightly less
+    # efficient as all processes with unexpected module names will wait on this lock
+    action_write_locks[None] = Lock()
+
+    # These plugins are known to be called directly by action plugins with names differing from the
+    # action plugin name.  We precreate them here as an optimization.
+    # If a list of service managers is created in the future we can do the same for them.
+    mods = set(p['name'] for p in PKG_MGRS)
+
+    mods.update(('copy', 'file', 'setup', 'slurp', 'stat'))
+    for mod_name in mods:
+        action_write_locks[mod_name] = Lock()
diff --git a/lib/ansible/executor/discovery/__init__.py b/lib/ansible/executor/discovery/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/lib/ansible/executor/discovery/__init__.py
diff --git a/lib/ansible/executor/discovery/python_target.py b/lib/ansible/executor/discovery/python_target.py
new file mode 100644
index 0000000..7137733
--- /dev/null
+++ b/lib/ansible/executor/discovery/python_target.py
@@ -0,0 +1,48 @@
+# Copyright: (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# FUTURE: this could be swapped out for our bundled version of distro to move more complete platform
+# logic to the targets, so long as we maintain Py2.6 compat and don't need to do any kind of script assembly
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import json
+import platform
+import io
+import os
+
+
+def read_utf8_file(path, encoding='utf-8'):
+    if not os.access(path, os.R_OK):
+        return None
+    with io.open(path, 'r', encoding=encoding) as fd:
+        content = fd.read()
+
+    return content
+
+
+def get_platform_info():
+    result = dict(platform_dist_result=[])
+
+    if hasattr(platform, 'dist'):
+        result['platform_dist_result'] = platform.dist()
+
+    osrelease_content = read_utf8_file('/etc/os-release')
+    # try to fall back to /usr/lib/os-release
+    if not osrelease_content:
+        osrelease_content = read_utf8_file('/usr/lib/os-release')
+
+    result['osrelease_content'] = osrelease_content
+
+    return result
+
+
+def main():
+    info = get_platform_info()
+
+    print(json.dumps(info))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/lib/ansible/executor/interpreter_discovery.py b/lib/ansible/executor/interpreter_discovery.py
new file mode 100644
index 0000000..bfd8504
--- /dev/null
+++ b/lib/ansible/executor/interpreter_discovery.py
@@ -0,0 +1,207 @@
+# Copyright: (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import bisect
+import json
+import pkgutil
+import re
+
+from ansible import constants as C
+from ansible.module_utils._text import to_native, to_text
+from ansible.module_utils.distro import LinuxDistribution
+from ansible.utils.display import Display
+from ansible.utils.plugin_docs import get_versioned_doclink
+from ansible.module_utils.compat.version import LooseVersion
+from ansible.module_utils.facts.system.distribution import Distribution
+from traceback import format_exc
+
+OS_FAMILY_LOWER = {k.lower(): v.lower() for k, v in Distribution.OS_FAMILY.items()}
+
+display = Display()
+foundre = re.compile(r'(?s)PLATFORM[\r\n]+(.*)FOUND(.*)ENDFOUND')
+
+
+class InterpreterDiscoveryRequiredError(Exception):
+    def __init__(self, message, interpreter_name, discovery_mode):
+        super(InterpreterDiscoveryRequiredError, self).__init__(message)
+        self.interpreter_name = interpreter_name
+        self.discovery_mode = discovery_mode
+
+    def __str__(self):
+        return self.message
+
+    def __repr__(self):
+        # TODO: proper repr impl
+        return self.message
+
+
+def discover_interpreter(action, interpreter_name, discovery_mode, task_vars):
+    # interpreter discovery is a 2-step process with the target. First, we use a simple shell-agnostic bootstrap to
+    # get the system type from uname, and find any random Python that can get us the info we need. For supported
+    # target OS types, we'll dispatch a Python script that calls plaform.dist() (for older platforms, where available)
+    # and brings back /etc/os-release (if present). The proper Python path is looked up in a table of known
+    # distros/versions with included Pythons; if nothing is found, depending on the discovery mode, either the
+    # default fallback of /usr/bin/python is used (if we know it's there), or discovery fails.
+
+    # FUTURE: add logical equivalence for "python3" in the case of py3-only modules?
+    if interpreter_name != 'python':
+        raise ValueError('Interpreter discovery not supported for {0}'.format(interpreter_name))
+
+    host = task_vars.get('inventory_hostname', 'unknown')
+    res = None
+    platform_type = 'unknown'
+    found_interpreters = [u'/usr/bin/python']  # fallback value
+    is_auto_legacy = discovery_mode.startswith('auto_legacy')
+    is_silent = discovery_mode.endswith('_silent')
+
+    try:
+        platform_python_map = C.config.get_config_value('_INTERPRETER_PYTHON_DISTRO_MAP', variables=task_vars)
+        bootstrap_python_list = C.config.get_config_value('INTERPRETER_PYTHON_FALLBACK', variables=task_vars)
+
+        display.vvv(msg=u"Attempting {0} interpreter discovery".format(interpreter_name), host=host)
+
+        # not all command -v impls accept a list of commands, so we have to call it once per python
+        command_list = ["command -v '%s'" % py for py in bootstrap_python_list]
+        shell_bootstrap = "echo PLATFORM; uname; echo FOUND; {0}; echo ENDFOUND".format('; '.join(command_list))
+
+        # FUTURE: in most cases we probably don't want to use become, but maybe sometimes we do?
+        res = action._low_level_execute_command(shell_bootstrap, sudoable=False)
+
+        raw_stdout = res.get('stdout', u'')
+
+        match = foundre.match(raw_stdout)
+
+        if not match:
+            display.debug(u'raw interpreter discovery output: {0}'.format(raw_stdout), host=host)
+            raise ValueError('unexpected output from Python interpreter discovery')
+
+        platform_type = match.groups()[0].lower().strip()
+
+        found_interpreters = [interp.strip() for interp in match.groups()[1].splitlines() if interp.startswith('/')]
+
+        display.debug(u"found interpreters: {0}".format(found_interpreters), host=host)
+
+        if not found_interpreters:
+            if not is_silent:
+                action._discovery_warnings.append(u'No python interpreters found for '
+                                                  u'host {0} (tried {1})'.format(host, bootstrap_python_list))
+            # this is lame, but returning None or throwing an exception is uglier
+            return u'/usr/bin/python'
+
+        if platform_type != 'linux':
+            raise NotImplementedError('unsupported platform for extended discovery: {0}'.format(to_native(platform_type)))
+
+        platform_script = pkgutil.get_data('ansible.executor.discovery', 'python_target.py')
+
+        # FUTURE: respect pipelining setting instead of just if the connection supports it?
+        if action._connection.has_pipelining:
+            res = action._low_level_execute_command(found_interpreters[0], sudoable=False, in_data=platform_script)
+        else:
+            # FUTURE: implement on-disk case (via script action or ?)
+            raise NotImplementedError('pipelining support required for extended interpreter discovery')
+
+        platform_info = json.loads(res.get('stdout'))
+
+        distro, version = _get_linux_distro(platform_info)
+
+        if not distro or not version:
+            raise NotImplementedError('unable to get Linux distribution/version info')
+
+        family = OS_FAMILY_LOWER.get(distro.lower().strip())
+
+        version_map = platform_python_map.get(distro.lower().strip()) or platform_python_map.get(family)
+        if not version_map:
+            raise NotImplementedError('unsupported Linux distribution: {0}'.format(distro))
+
+        platform_interpreter = to_text(_version_fuzzy_match(version, version_map), errors='surrogate_or_strict')
+
+        # provide a transition period for hosts that were using /usr/bin/python previously (but shouldn't have been)
+        if is_auto_legacy:
+            if platform_interpreter != u'/usr/bin/python' and u'/usr/bin/python' in found_interpreters:
+                if not is_silent:
+                    action._discovery_warnings.append(
+                        u"Distribution {0} {1} on host {2} should use {3}, but is using "
+                        u"/usr/bin/python for backward compatibility with prior Ansible releases. "
+                        u"See {4} for more information"
+                        .format(distro, version, host, platform_interpreter,
+                                get_versioned_doclink('reference_appendices/interpreter_discovery.html')))
+                return u'/usr/bin/python'
+
+        if platform_interpreter not in found_interpreters:
+            if platform_interpreter not in bootstrap_python_list:
+                # sanity check to make sure we looked for it
+                if not is_silent:
+                    action._discovery_warnings \
+                        .append(u"Platform interpreter {0} on host {1} is missing from bootstrap list"
+                                .format(platform_interpreter, host))
+
+            if not is_silent:
+                action._discovery_warnings \
+                    .append(u"Distribution {0} {1} on host {2} should use {3}, but is using {4}, since the "
+                            u"discovered platform python interpreter was not present. See {5} "
+                            u"for more information."
+                            .format(distro, version, host, platform_interpreter, found_interpreters[0],
+                                    get_versioned_doclink('reference_appendices/interpreter_discovery.html')))
+            return found_interpreters[0]
+
+        return platform_interpreter
+    except NotImplementedError as ex:
+        display.vvv(msg=u'Python interpreter discovery fallback ({0})'.format(to_text(ex)), host=host)
+    except Exception as ex:
+        if not is_silent:
+            display.warning(msg=u'Unhandled error in Python interpreter discovery for host {0}: {1}'.format(host, to_text(ex)))
+            display.debug(msg=u'Interpreter discovery traceback:\n{0}'.format(to_text(format_exc())), host=host)
+            if res and res.get('stderr'):
+                display.vvv(msg=u'Interpreter discovery remote stderr:\n{0}'.format(to_text(res.get('stderr'))), host=host)
+
+    if not is_silent:
+        action._discovery_warnings \
+            .append(u"Platform {0} on host {1} is using the discovered Python interpreter at {2}, but future installation of "
+                    u"another Python interpreter could change the meaning of that path. See {3} "
+                    u"for more information."
+                    .format(platform_type, host, found_interpreters[0],
+                            get_versioned_doclink('reference_appendices/interpreter_discovery.html')))
+    return found_interpreters[0]
+
+
+def _get_linux_distro(platform_info):
+    dist_result = platform_info.get('platform_dist_result', [])
+
+    if len(dist_result) == 3 and any(dist_result):
+        return dist_result[0], dist_result[1]
+
+    osrelease_content = platform_info.get('osrelease_content')
+
+    if not osrelease_content:
+        return u'', u''
+
+    osr = LinuxDistribution._parse_os_release_content(osrelease_content)
+
+    return osr.get('id', u''), osr.get('version_id', u'')
+
+
+def _version_fuzzy_match(version, version_map):
+    # try exact match first
+    res = version_map.get(version)
+    if res:
+        return res
+
+    sorted_looseversions = sorted([LooseVersion(v) for v in version_map.keys()])
+
+    find_looseversion = LooseVersion(version)
+
+    # slot match; return nearest previous version we're newer than
+    kpos = bisect.bisect(sorted_looseversions, find_looseversion)
+
+    if kpos == 0:
+        # older than everything in the list, return the oldest version
+        # TODO: warning-worthy?
+        return version_map.get(sorted_looseversions[0].vstring)
+
+    # TODO: is "past the end of the list" warning-worthy too (at least if it's not a major version match)?
+
+    # return the next-oldest entry that we're newer than...
+    return version_map.get(sorted_looseversions[kpos - 1].vstring)
diff --git a/lib/ansible/executor/module_common.py b/lib/ansible/executor/module_common.py
new file mode 100644
index 0000000..4d06acb
--- /dev/null
+++ b/lib/ansible/executor/module_common.py
@@ -0,0 +1,1428 @@
+# (c) 2013-2014, Michael DeHaan <michael.dehaan@gmail.com>
+# (c) 2015 Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import ast
+import base64
+import datetime
+import json
+import os
+import shlex
+import zipfile
+import re
+import pkgutil
+
+from ast import AST, Import, ImportFrom
+from io import BytesIO
+
+from ansible.release import __version__, __author__
+from ansible import constants as C
+from ansible.errors import AnsibleError
+from ansible.executor.interpreter_discovery import InterpreterDiscoveryRequiredError
+from ansible.executor.powershell import module_manifest as ps_manifest
+from ansible.module_utils.common.json import AnsibleJSONEncoder
+from ansible.module_utils.common.text.converters import to_bytes, to_text, to_native
+from ansible.plugins.loader import module_utils_loader
+from ansible.utils.collection_loader._collection_finder import _get_collection_metadata, _nested_dict_get
+
+# Must import strategy and use write_locks from there
+# If we import write_locks directly then we end up binding a
+# variable to the object and then it never gets updated.
+from ansible.executor import action_write_locks
+
+from ansible.utils.display import Display
+from collections import namedtuple
+
+import importlib.util
+import importlib.machinery
+
+display = Display()
+
+ModuleUtilsProcessEntry = namedtuple('ModuleUtilsProcessEntry', ['name_parts', 'is_ambiguous', 'has_redirected_child', 'is_optional'])
+
+REPLACER = b"#<<INCLUDE_ANSIBLE_MODULE_COMMON>>"
+REPLACER_VERSION = b"\"<<ANSIBLE_VERSION>>\""
+REPLACER_COMPLEX = b"\"<<INCLUDE_ANSIBLE_MODULE_COMPLEX_ARGS>>\""
+REPLACER_WINDOWS = b"# POWERSHELL_COMMON"
+REPLACER_JSONARGS = b"<<INCLUDE_ANSIBLE_MODULE_JSON_ARGS>>"
+REPLACER_SELINUX = b"<<SELINUX_SPECIAL_FILESYSTEMS>>"
+
+# We could end up writing out parameters with unicode characters so we need to
+# specify an encoding for the python source file
+ENCODING_STRING = u'# -*- coding: utf-8 -*-'
+b_ENCODING_STRING = b'# -*- coding: utf-8 -*-'
+
+# module_common is relative to module_utils, so fix the path
+_MODULE_UTILS_PATH = os.path.join(os.path.dirname(__file__), '..', 'module_utils')
+
+# ******************************************************************************
+
+ANSIBALLZ_TEMPLATE = u'''%(shebang)s
+%(coding)s
+_ANSIBALLZ_WRAPPER = True # For test-module.py script to tell this is a ANSIBALLZ_WRAPPER
+# This code is part of Ansible, but is an independent component.
+# The code in this particular templatable string, and this templatable string
+# only, is BSD licensed.  Modules which end up using this snippet, which is
+# dynamically combined together by Ansible still belong to the author of the
+# module, and they may assign their own license to the complete work.
+#
+# Copyright (c), James Cammarata, 2016
+# Copyright (c), Toshio Kuratomi, 2016
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+#
+#    * Redistributions of source code must retain the above copyright
+#      notice, this list of conditions and the following disclaimer.
+#    * Redistributions in binary form must reproduce the above copyright notice,
+#      this list of conditions and the following disclaimer in the documentation
+#      and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+# IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+# USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+def _ansiballz_main():
+    import os
+    import os.path
+
+    # Access to the working directory is required by Python when using pipelining, as well as for the coverage module.
+    # Some platforms, such as macOS, may not allow querying the working directory when using become to drop privileges.
+    try:
+        os.getcwd()
+    except OSError:
+        try:
+            os.chdir(os.path.expanduser('~'))
+        except OSError:
+            os.chdir('/')
+
+%(rlimit)s
+
+    import sys
+    import __main__
+
+    # For some distros and python versions we pick up this script in the temporary
+    # directory.  This leads to problems when the ansible module masks a python
+    # library that another import needs.  We have not figured out what about the
+    # specific distros and python versions causes this to behave differently.
+    #
+    # Tested distros:
+    # Fedora23 with python3.4  Works
+    # Ubuntu15.10 with python2.7  Works
+    # Ubuntu15.10 with python3.4  Fails without this
+    # Ubuntu16.04.1 with python3.5  Fails without this
+    # To test on another platform:
+    # * use the copy module (since this shadows the stdlib copy module)
+    # * Turn off pipelining
+    # * Make sure that the destination file does not exist
+    # * ansible ubuntu16-test -m copy -a 'src=/etc/motd dest=/var/tmp/m'
+    # This will traceback in shutil.  Looking at the complete traceback will show
+    # that shutil is importing copy which finds the ansible module instead of the
+    # stdlib module
+    scriptdir = None
+    try:
+        scriptdir = os.path.dirname(os.path.realpath(__main__.__file__))
+    except (AttributeError, OSError):
+        # Some platforms don't set __file__ when reading from stdin
+        # OSX raises OSError if using abspath() in a directory we don't have
+        # permission to read (realpath calls abspath)
+        pass
+
+    # Strip cwd from sys.path to avoid potential permissions issues
+    excludes = set(('', '.', scriptdir))
+    sys.path = [p for p in sys.path if p not in excludes]
+
+    import base64
+    import runpy
+    import shutil
+    import tempfile
+    import zipfile
+
+    if sys.version_info < (3,):
+        PY3 = False
+    else:
+        PY3 = True
+
+    ZIPDATA = """%(zipdata)s"""
+
+    # Note: temp_path isn't needed once we switch to zipimport
+    def invoke_module(modlib_path, temp_path, json_params):
+        # When installed via setuptools (including python setup.py install),
+        # ansible may be installed with an easy-install.pth file.  That file
+        # may load the system-wide install of ansible rather than the one in
+        # the module.  sitecustomize is the only way to override that setting.
+        z = zipfile.ZipFile(modlib_path, mode='a')
+
+        # py3: modlib_path will be text, py2: it's bytes.  Need bytes at the end
+        sitecustomize = u'import sys\\nsys.path.insert(0,"%%s")\\n' %%  modlib_path
+        sitecustomize = sitecustomize.encode('utf-8')
+        # Use a ZipInfo to work around zipfile limitation on hosts with
+        # clocks set to a pre-1980 year (for instance, Raspberry Pi)
+        zinfo = zipfile.ZipInfo()
+        zinfo.filename = 'sitecustomize.py'
+        zinfo.date_time = ( %(year)i, %(month)i, %(day)i, %(hour)i, %(minute)i, %(second)i)
+        z.writestr(zinfo, sitecustomize)
+        z.close()
+
+        # Put the zipped up module_utils we got from the controller first in the python path so that we
+        # can monkeypatch the right basic
+        sys.path.insert(0, modlib_path)
+
+        # Monkeypatch the parameters into basic
+        from ansible.module_utils import basic
+        basic._ANSIBLE_ARGS = json_params
+%(coverage)s
+        # Run the module!  By importing it as '__main__', it thinks it is executing as a script
+        runpy.run_module(mod_name='%(module_fqn)s', init_globals=dict(_module_fqn='%(module_fqn)s', _modlib_path=modlib_path),
+                         run_name='__main__', alter_sys=True)
+
+        # Ansible modules must exit themselves
+        print('{"msg": "New-style module did not handle its own exit", "failed": true}')
+        sys.exit(1)
+
+    def debug(command, zipped_mod, json_params):
+        # The code here normally doesn't run.  It's only used for debugging on the
+        # remote machine.
+        #
+        # The subcommands in this function make it easier to debug ansiballz
+        # modules.  Here's the basic steps:
+        #
+        # Run ansible with the environment variable: ANSIBLE_KEEP_REMOTE_FILES=1 and -vvv
+        # to save the module file remotely::
+        #   $ ANSIBLE_KEEP_REMOTE_FILES=1 ansible host1 -m ping -a 'data=october' -vvv
+        #
+        # Part of the verbose output will tell you where on the remote machine the
+        # module was written to::
+        #   [...]
+        #   <host1> SSH: EXEC ssh -C -q -o ControlMaster=auto -o ControlPersist=60s -o KbdInteractiveAuthentication=no -o
+        #   PreferredAuthentications=gssapi-with-mic,gssapi-keyex,hostbased,publickey -o PasswordAuthentication=no -o ConnectTimeout=10 -o
+        #   ControlPath=/home/badger/.ansible/cp/ansible-ssh-%%h-%%p-%%r -tt rhel7 '/bin/sh -c '"'"'LANG=en_US.UTF-8 LC_ALL=en_US.UTF-8
+        #   LC_MESSAGES=en_US.UTF-8 /usr/bin/python /home/badger/.ansible/tmp/ansible-tmp-1461173013.93-9076457629738/ping'"'"''
+        #   [...]
+        #
+        # Login to the remote machine and run the module file via from the previous
+        # step with the explode subcommand to extract the module payload into
+        # source files::
+        #   $ ssh host1
+        #   $ /usr/bin/python /home/badger/.ansible/tmp/ansible-tmp-1461173013.93-9076457629738/ping explode
+        #   Module expanded into:
+        #   /home/badger/.ansible/tmp/ansible-tmp-1461173408.08-279692652635227/ansible
+        #
+        # You can now edit the source files to instrument the code or experiment with
+        # different parameter values.  When you're ready to run the code you've modified
+        # (instead of the code from the actual zipped module), use the execute subcommand like this::
+        #   $ /usr/bin/python /home/badger/.ansible/tmp/ansible-tmp-1461173013.93-9076457629738/ping execute
+
+        # Okay to use __file__ here because we're running from a kept file
+        basedir = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'debug_dir')
+        args_path = os.path.join(basedir, 'args')
+
+        if command == 'explode':
+            # transform the ZIPDATA into an exploded directory of code and then
+            # print the path to the code.  This is an easy way for people to look
+            # at the code on the remote machine for debugging it in that
+            # environment
+            z = zipfile.ZipFile(zipped_mod)
+            for filename in z.namelist():
+                if filename.startswith('/'):
+                    raise Exception('Something wrong with this module zip file: should not contain absolute paths')
+
+                dest_filename = os.path.join(basedir, filename)
+                if dest_filename.endswith(os.path.sep) and not os.path.exists(dest_filename):
+                    os.makedirs(dest_filename)
+                else:
+                    directory = os.path.dirname(dest_filename)
+                    if not os.path.exists(directory):
+                        os.makedirs(directory)
+                    f = open(dest_filename, 'wb')
+                    f.write(z.read(filename))
+                    f.close()
+
+            # write the args file
+            f = open(args_path, 'wb')
+            f.write(json_params)
+            f.close()
+
+            print('Module expanded into:')
+            print('%%s' %% basedir)
+            exitcode = 0
+
+        elif command == 'execute':
+            # Execute the exploded code instead of executing the module from the
+            # embedded ZIPDATA.  This allows people to easily run their modified
+            # code on the remote machine to see how changes will affect it.
+
+            # Set pythonpath to the debug dir
+            sys.path.insert(0, basedir)
+
+            # read in the args file which the user may have modified
+            with open(args_path, 'rb') as f:
+                json_params = f.read()
+
+            # Monkeypatch the parameters into basic
+            from ansible.module_utils import basic
+            basic._ANSIBLE_ARGS = json_params
+
+            # Run the module!  By importing it as '__main__', it thinks it is executing as a script
+            runpy.run_module(mod_name='%(module_fqn)s', init_globals=None, run_name='__main__', alter_sys=True)
+
+            # Ansible modules must exit themselves
+            print('{"msg": "New-style module did not handle its own exit", "failed": true}')
+            sys.exit(1)
+
+        else:
+            print('WARNING: Unknown debug command.  Doing nothing.')
+            exitcode = 0
+
+        return exitcode
+
+    #
+    # See comments in the debug() method for information on debugging
+    #
+
+    ANSIBALLZ_PARAMS = %(params)s
+    if PY3:
+        ANSIBALLZ_PARAMS = ANSIBALLZ_PARAMS.encode('utf-8')
+    try:
+        # There's a race condition with the controller removing the
+        # remote_tmpdir and this module executing under async.  So we cannot
+        # store this in remote_tmpdir (use system tempdir instead)
+        # Only need to use [ansible_module]_payload_ in the temp_path until we move to zipimport
+        # (this helps ansible-test produce coverage stats)
+        temp_path = tempfile.mkdtemp(prefix='ansible_%(ansible_module)s_payload_')
+
+        zipped_mod = os.path.join(temp_path, 'ansible_%(ansible_module)s_payload.zip')
+
+        with open(zipped_mod, 'wb') as modlib:
+            modlib.write(base64.b64decode(ZIPDATA))
+
+        if len(sys.argv) == 2:
+            exitcode = debug(sys.argv[1], zipped_mod, ANSIBALLZ_PARAMS)
+        else:
+            # Note: temp_path isn't needed once we switch to zipimport
+            invoke_module(zipped_mod, temp_path, ANSIBALLZ_PARAMS)
+    finally:
+        try:
+            shutil.rmtree(temp_path)
+        except (NameError, OSError):
+            # tempdir creation probably failed
+            pass
+    sys.exit(exitcode)
+
+if __name__ == '__main__':
+    _ansiballz_main()
+'''
+
+ANSIBALLZ_COVERAGE_TEMPLATE = '''
+        os.environ['COVERAGE_FILE'] = '%(coverage_output)s=python-%%s=coverage' %% '.'.join(str(v) for v in sys.version_info[:2])
+
+        import atexit
+
+        try:
+            import coverage
+        except ImportError:
+            print('{"msg": "Could not import `coverage` module.", "failed": true}')
+            sys.exit(1)
+
+        cov = coverage.Coverage(config_file='%(coverage_config)s')
+
+        def atexit_coverage():
+            cov.stop()
+            cov.save()
+
+        atexit.register(atexit_coverage)
+
+        cov.start()
+'''
+
+ANSIBALLZ_COVERAGE_CHECK_TEMPLATE = '''
+        try:
+            if PY3:
+                import importlib.util
+                if importlib.util.find_spec('coverage') is None:
+                    raise ImportError
+            else:
+                import imp
+                imp.find_module('coverage')
+        except ImportError:
+            print('{"msg": "Could not find `coverage` module.", "failed": true}')
+            sys.exit(1)
+'''
+
+ANSIBALLZ_RLIMIT_TEMPLATE = '''
+    import resource
+
+    existing_soft, existing_hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+
+    # adjust soft limit subject to existing hard limit
+    requested_soft = min(existing_hard, %(rlimit_nofile)d)
+
+    if requested_soft != existing_soft:
+        try:
+            resource.setrlimit(resource.RLIMIT_NOFILE, (requested_soft, existing_hard))
+        except ValueError:
+            # some platforms (eg macOS) lie about their hard limit
+            pass
+'''
+
+
+def _strip_comments(source):
+    # Strip comments and blank lines from the wrapper
+    buf = []
+    for line in source.splitlines():
+        l = line.strip()
+        if not l or l.startswith(u'#'):
+            continue
+        buf.append(line)
+    return u'\n'.join(buf)
+
+
+if C.DEFAULT_KEEP_REMOTE_FILES:
+    # Keep comments when KEEP_REMOTE_FILES is set.  That way users will see
+    # the comments with some nice usage instructions
+    ACTIVE_ANSIBALLZ_TEMPLATE = ANSIBALLZ_TEMPLATE
+else:
+    # ANSIBALLZ_TEMPLATE stripped of comments for smaller over the wire size
+    ACTIVE_ANSIBALLZ_TEMPLATE = _strip_comments(ANSIBALLZ_TEMPLATE)
+
+# dirname(dirname(dirname(site-packages/ansible/executor/module_common.py) == site-packages
+# Do this instead of getting site-packages from distutils.sysconfig so we work when we
+# haven't been installed
+site_packages = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+CORE_LIBRARY_PATH_RE = re.compile(r'%s/(?P<path>ansible/modules/.*)\.(py|ps1)$' % re.escape(site_packages))
+COLLECTION_PATH_RE = re.compile(r'/(?P<path>ansible_collections/[^/]+/[^/]+/plugins/modules/.*)\.(py|ps1)$')
+
+# Detect new-style Python modules by looking for required imports:
+# import ansible_collections.[my_ns.my_col.plugins.module_utils.my_module_util]
+# from ansible_collections.[my_ns.my_col.plugins.module_utils import my_module_util]
+# import ansible.module_utils[.basic]
+# from ansible.module_utils[ import basic]
+# from ansible.module_utils[.basic import AnsibleModule]
+# from ..module_utils[ import basic]
+# from ..module_utils[.basic import AnsibleModule]
+NEW_STYLE_PYTHON_MODULE_RE = re.compile(
+    # Relative imports
+    br'(?:from +\.{2,} *module_utils.* +import |'
+    # Collection absolute imports:
+    br'from +ansible_collections\.[^.]+\.[^.]+\.plugins\.module_utils.* +import |'
+    br'import +ansible_collections\.[^.]+\.[^.]+\.plugins\.module_utils.*|'
+    # Core absolute imports
+    br'from +ansible\.module_utils.* +import |'
+    br'import +ansible\.module_utils\.)'
+)
+
+
+class ModuleDepFinder(ast.NodeVisitor):
+    def __init__(self, module_fqn, tree, is_pkg_init=False, *args, **kwargs):
+        """
+        Walk the ast tree for the python module.
+        :arg module_fqn: The fully qualified name to reach this module in dotted notation.
+            example: ansible.module_utils.basic
+        :arg is_pkg_init: Inform the finder it's looking at a package init (eg __init__.py) to allow
+            relative import expansion to use the proper package level without having imported it locally first.
+
+        Save submodule[.submoduleN][.identifier] into self.submodules
+        when they are from ansible.module_utils or ansible_collections packages
+
+        self.submodules will end up with tuples like:
+          - ('ansible', 'module_utils', 'basic',)
+          - ('ansible', 'module_utils', 'urls', 'fetch_url')
+          - ('ansible', 'module_utils', 'database', 'postgres')
+          - ('ansible', 'module_utils', 'database', 'postgres', 'quote')
+          - ('ansible', 'module_utils', 'database', 'postgres', 'quote')
+          - ('ansible_collections', 'my_ns', 'my_col', 'plugins', 'module_utils', 'foo')
+
+        It's up to calling code to determine whether the final element of the
+        tuple are module names or something else (function, class, or variable names)
+        .. seealso:: :python3:class:`ast.NodeVisitor`
+        """
+        super(ModuleDepFinder, self).__init__(*args, **kwargs)
+        self._tree = tree  # squirrel this away so we can compare node parents to it
+        self.submodules = set()
+        self.optional_imports = set()
+        self.module_fqn = module_fqn
+        self.is_pkg_init = is_pkg_init
+
+        self._visit_map = {
+            Import: self.visit_Import,
+            ImportFrom: self.visit_ImportFrom,
+        }
+
+        self.visit(tree)
+
+    def generic_visit(self, node):
+        """Overridden ``generic_visit`` that makes some assumptions about our
+        use case, and improves performance by calling visitors directly instead
+        of calling ``visit`` to offload calling visitors.
+        """
+        generic_visit = self.generic_visit
+        visit_map = self._visit_map
+        for field, value in ast.iter_fields(node):
+            if isinstance(value, list):
+                for item in value:
+                    if isinstance(item, (Import, ImportFrom)):
+                        item.parent = node
+                        visit_map[item.__class__](item)
+                    elif isinstance(item, AST):
+                        generic_visit(item)
+
+    visit = generic_visit
+
+    def visit_Import(self, node):
+        """
+        Handle import ansible.module_utils.MODLIB[.MODLIBn] [as asname]
+
+        We save these as interesting submodules when the imported library is in ansible.module_utils
+        or ansible.collections
+        """
+        for alias in node.names:
+            if (alias.name.startswith('ansible.module_utils.') or
+                    alias.name.startswith('ansible_collections.')):
+                py_mod = tuple(alias.name.split('.'))
+                self.submodules.add(py_mod)
+                # if the import's parent is the root document, it's a required import, otherwise it's optional
+                if node.parent != self._tree:
+                    self.optional_imports.add(py_mod)
+        self.generic_visit(node)
+
+    def visit_ImportFrom(self, node):
+        """
+        Handle from ansible.module_utils.MODLIB import [.MODLIBn] [as asname]
+
+        Also has to handle relative imports
+
+        We save these as interesting submodules when the imported library is in ansible.module_utils
+        or ansible.collections
+        """
+
+        # FIXME: These should all get skipped:
+        # from ansible.executor import module_common
+        # from ...executor import module_common
+        # from ... import executor (Currently it gives a non-helpful error)
+        if node.level > 0:
+            # if we're in a package init, we have to add one to the node level (and make it none if 0 to preserve the right slicing behavior)
+            level_slice_offset = -node.level + 1 or None if self.is_pkg_init else -node.level
+            if self.module_fqn:
+                parts = tuple(self.module_fqn.split('.'))
+                if node.module:
+                    # relative import: from .module import x
+                    node_module = '.'.join(parts[:level_slice_offset] + (node.module,))
+                else:
+                    # relative import: from . import x
+                    node_module = '.'.join(parts[:level_slice_offset])
+            else:
+                # fall back to an absolute import
+                node_module = node.module
+        else:
+            # absolute import: from module import x
+            node_module = node.module
+
+        # Specialcase: six is a special case because of its
+        # import logic
+        py_mod = None
+        if node.names[0].name == '_six':
+            self.submodules.add(('_six',))
+        elif node_module.startswith('ansible.module_utils'):
+            # from ansible.module_utils.MODULE1[.MODULEn] import IDENTIFIER [as asname]
+            # from ansible.module_utils.MODULE1[.MODULEn] import MODULEn+1 [as asname]
+            # from ansible.module_utils.MODULE1[.MODULEn] import MODULEn+1 [,IDENTIFIER] [as asname]
+            # from ansible.module_utils import MODULE1 [,MODULEn] [as asname]
+            py_mod = tuple(node_module.split('.'))
+
+        elif node_module.startswith('ansible_collections.'):
+            if node_module.endswith('plugins.module_utils') or '.plugins.module_utils.' in node_module:
+                # from ansible_collections.ns.coll.plugins.module_utils import MODULE [as aname] [,MODULE2] [as aname]
+                # from ansible_collections.ns.coll.plugins.module_utils.MODULE import IDENTIFIER [as aname]
+                # FIXME: Unhandled cornercase (needs to be ignored):
+                # from ansible_collections.ns.coll.plugins.[!module_utils].[FOO].plugins.module_utils import IDENTIFIER
+                py_mod = tuple(node_module.split('.'))
+            else:
+                # Not from module_utils so ignore.  for instance:
+                # from ansible_collections.ns.coll.plugins.lookup import IDENTIFIER
+                pass
+
+        if py_mod:
+            for alias in node.names:
+                self.submodules.add(py_mod + (alias.name,))
+                # if the import's parent is the root document, it's a required import, otherwise it's optional
+                if node.parent != self._tree:
+                    self.optional_imports.add(py_mod + (alias.name,))
+
+        self.generic_visit(node)
+
+
+def _slurp(path):
+    if not os.path.exists(path):
+        raise AnsibleError("imported module support code does not exist at %s" % os.path.abspath(path))
+    with open(path, 'rb') as fd:
+        data = fd.read()
+    return data
+
+
+def _get_shebang(interpreter, task_vars, templar, args=tuple(), remote_is_local=False):
+    """
+      Handles the different ways ansible allows overriding the shebang target for a module.
+    """
+    # FUTURE: add logical equivalence for python3 in the case of py3-only modules
+
+    interpreter_name = os.path.basename(interpreter).strip()
+
+    # name for interpreter var
+    interpreter_config = u'ansible_%s_interpreter' % interpreter_name
+    # key for config
+    interpreter_config_key = "INTERPRETER_%s" % interpreter_name.upper()
+
+    interpreter_out = None
+
+    # looking for python, rest rely on matching vars
+    if interpreter_name == 'python':
+        # skip detection for network os execution, use playbook supplied one if possible
+        if remote_is_local:
+            interpreter_out = task_vars['ansible_playbook_python']
+
+        # a config def exists for this interpreter type; consult config for the value
+        elif C.config.get_configuration_definition(interpreter_config_key):
+
+            interpreter_from_config = C.config.get_config_value(interpreter_config_key, variables=task_vars)
+            interpreter_out = templar.template(interpreter_from_config.strip())
+
+            # handle interpreter discovery if requested or empty interpreter was provided
+            if not interpreter_out or interpreter_out in ['auto', 'auto_legacy', 'auto_silent', 'auto_legacy_silent']:
+
+                discovered_interpreter_config = u'discovered_interpreter_%s' % interpreter_name
+                facts_from_task_vars = task_vars.get('ansible_facts', {})
+
+                if discovered_interpreter_config not in facts_from_task_vars:
+                    # interpreter discovery is desired, but has not been run for this host
+                    raise InterpreterDiscoveryRequiredError("interpreter discovery needed", interpreter_name=interpreter_name, discovery_mode=interpreter_out)
+                else:
+                    interpreter_out = facts_from_task_vars[discovered_interpreter_config]
+        else:
+            raise InterpreterDiscoveryRequiredError("interpreter discovery required", interpreter_name=interpreter_name, discovery_mode='auto_legacy')
+
+    elif interpreter_config in task_vars:
+        # for non python we consult vars for a possible direct override
+        interpreter_out = templar.template(task_vars.get(interpreter_config).strip())
+
+    if not interpreter_out:
+        # nothing matched(None) or in case someone configures empty string or empty intepreter
+        interpreter_out = interpreter
+
+    # set shebang
+    shebang = u'#!{0}'.format(interpreter_out)
+    if args:
+        shebang = shebang + u' ' + u' '.join(args)
+
+    return shebang, interpreter_out
+
+
+class ModuleUtilLocatorBase:
+    def __init__(self, fq_name_parts, is_ambiguous=False, child_is_redirected=False, is_optional=False):
+        self._is_ambiguous = is_ambiguous
+        # a child package redirection could cause intermediate package levels to be missing, eg
+        # from ansible.module_utils.x.y.z import foo; if x.y.z.foo is redirected, we may not have packages on disk for
+        # the intermediate packages x.y.z, so we'll need to supply empty packages for those
+        self._child_is_redirected = child_is_redirected
+        self._is_optional = is_optional
+        self.found = False
+        self.redirected = False
+        self.fq_name_parts = fq_name_parts
+        self.source_code = ''
+        self.output_path = ''
+        self.is_package = False
+        self._collection_name = None
+        # for ambiguous imports, we should only test for things more than one level below module_utils
+        # this lets us detect erroneous imports and redirections earlier
+        if is_ambiguous and len(self._get_module_utils_remainder_parts(fq_name_parts)) > 1:
+            self.candidate_names = [fq_name_parts, fq_name_parts[:-1]]
+        else:
+            self.candidate_names = [fq_name_parts]
+
+    @property
+    def candidate_names_joined(self):
+        return ['.'.join(n) for n in self.candidate_names]
+
+    def _handle_redirect(self, name_parts):
+        module_utils_relative_parts = self._get_module_utils_remainder_parts(name_parts)
+
+        # only allow redirects from below module_utils- if above that, bail out (eg, parent package names)
+        if not module_utils_relative_parts:
+            return False
+
+        try:
+            collection_metadata = _get_collection_metadata(self._collection_name)
+        except ValueError as ve:  # collection not found or some other error related to collection load
+            if self._is_optional:
+                return False
+            raise AnsibleError('error processing module_util {0} loading redirected collection {1}: {2}'
+                               .format('.'.join(name_parts), self._collection_name, to_native(ve)))
+
+        routing_entry = _nested_dict_get(collection_metadata, ['plugin_routing', 'module_utils', '.'.join(module_utils_relative_parts)])
+        if not routing_entry:
+            return False
+        # FIXME: add deprecation warning support
+
+        dep_or_ts = routing_entry.get('tombstone')
+        removed = dep_or_ts is not None
+        if not removed:
+            dep_or_ts = routing_entry.get('deprecation')
+
+        if dep_or_ts:
+            removal_date = dep_or_ts.get('removal_date')
+            removal_version = dep_or_ts.get('removal_version')
+            warning_text = dep_or_ts.get('warning_text')
+
+            msg = 'module_util {0} has been removed'.format('.'.join(name_parts))
+            if warning_text:
+                msg += ' ({0})'.format(warning_text)
+            else:
+                msg += '.'
+
+            display.deprecated(msg, removal_version, removed, removal_date, self._collection_name)
+        if 'redirect' in routing_entry:
+            self.redirected = True
+            source_pkg = '.'.join(name_parts)
+            self.is_package = True  # treat all redirects as packages
+            redirect_target_pkg = routing_entry['redirect']
+
+            # expand FQCN redirects
+            if not redirect_target_pkg.startswith('ansible_collections'):
+                split_fqcn = redirect_target_pkg.split('.')
+                if len(split_fqcn) < 3:
+                    raise Exception('invalid redirect for {0}: {1}'.format(source_pkg, redirect_target_pkg))
+                # assume it's an FQCN, expand it
+                redirect_target_pkg = 'ansible_collections.{0}.{1}.plugins.module_utils.{2}'.format(
+                    split_fqcn[0],  # ns
+                    split_fqcn[1],  # coll
+                    '.'.join(split_fqcn[2:])  # sub-module_utils remainder
+                )
+            display.vvv('redirecting module_util {0} to {1}'.format(source_pkg, redirect_target_pkg))
+            self.source_code = self._generate_redirect_shim_source(source_pkg, redirect_target_pkg)
+            return True
+        return False
+
+    def _get_module_utils_remainder_parts(self, name_parts):
+        # subclasses should override to return the name parts after module_utils
+        return []
+
+    def _get_module_utils_remainder(self, name_parts):
+        # return the remainder parts as a package string
+        return '.'.join(self._get_module_utils_remainder_parts(name_parts))
+
+    def _find_module(self, name_parts):
+        return False
+
+    def _locate(self, redirect_first=True):
+        for candidate_name_parts in self.candidate_names:
+            if redirect_first and self._handle_redirect(candidate_name_parts):
+                break
+
+            if self._find_module(candidate_name_parts):
+                break
+
+            if not redirect_first and self._handle_redirect(candidate_name_parts):
+                break
+
+        else:  # didn't find what we were looking for- last chance for packages whose parents were redirected
+            if self._child_is_redirected:  # make fake packages
+                self.is_package = True
+                self.source_code = ''
+            else:  # nope, just bail
+                return
+
+        if self.is_package:
+            path_parts = candidate_name_parts + ('__init__',)
+        else:
+            path_parts = candidate_name_parts
+        self.found = True
+        self.output_path = os.path.join(*path_parts) + '.py'
+        self.fq_name_parts = candidate_name_parts
+
+    def _generate_redirect_shim_source(self, fq_source_module, fq_target_module):
+        return """
+import sys
+import {1} as mod
+
+sys.modules['{0}'] = mod
+""".format(fq_source_module, fq_target_module)
+
+        # FIXME: add __repr__ impl
+
+
+class LegacyModuleUtilLocator(ModuleUtilLocatorBase):
+    def __init__(self, fq_name_parts, is_ambiguous=False, mu_paths=None, child_is_redirected=False):
+        super(LegacyModuleUtilLocator, self).__init__(fq_name_parts, is_ambiguous, child_is_redirected)
+
+        if fq_name_parts[0:2] != ('ansible', 'module_utils'):
+            raise Exception('this class can only locate from ansible.module_utils, got {0}'.format(fq_name_parts))
+
+        if fq_name_parts[2] == 'six':
+            # FIXME: handle the ansible.module_utils.six._six case with a redirect or an internal _six attr on six itself?
+            # six creates its submodules at runtime; convert all these to just 'ansible.module_utils.six'
+            fq_name_parts = ('ansible', 'module_utils', 'six')
+            self.candidate_names = [fq_name_parts]
+
+        self._mu_paths = mu_paths
+        self._collection_name = 'ansible.builtin'  # legacy module utils always look in ansible.builtin for redirects
+        self._locate(redirect_first=False)  # let local stuff override redirects for legacy
+
+    def _get_module_utils_remainder_parts(self, name_parts):
+        return name_parts[2:]  # eg, foo.bar for ansible.module_utils.foo.bar
+
+    def _find_module(self, name_parts):
+        rel_name_parts = self._get_module_utils_remainder_parts(name_parts)
+
+        # no redirection; try to find the module
+        if len(rel_name_parts) == 1:  # direct child of module_utils, just search the top-level dirs we were given
+            paths = self._mu_paths
+        else:  # a nested submodule of module_utils, extend the paths given with the intermediate package names
+            paths = [os.path.join(p, *rel_name_parts[:-1]) for p in
+                     self._mu_paths]  # extend the MU paths with the relative bit
+
+        # find_spec needs the full module name
+        self._info = info = importlib.machinery.PathFinder.find_spec('.'.join(name_parts), paths)
+        if info is not None and os.path.splitext(info.origin)[1] in importlib.machinery.SOURCE_SUFFIXES:
+            self.is_package = info.origin.endswith('/__init__.py')
+            path = info.origin
+        else:
+            return False
+        self.source_code = _slurp(path)
+
+        return True
+
+
+class CollectionModuleUtilLocator(ModuleUtilLocatorBase):
+    def __init__(self, fq_name_parts, is_ambiguous=False, child_is_redirected=False, is_optional=False):
+        super(CollectionModuleUtilLocator, self).__init__(fq_name_parts, is_ambiguous, child_is_redirected, is_optional)
+
+        if fq_name_parts[0] != 'ansible_collections':
+            raise Exception('CollectionModuleUtilLocator can only locate from ansible_collections, got {0}'.format(fq_name_parts))
+        elif len(fq_name_parts) >= 6 and fq_name_parts[3:5] != ('plugins', 'module_utils'):
+            raise Exception('CollectionModuleUtilLocator can only locate below ansible_collections.(ns).(coll).plugins.module_utils, got {0}'
+                            .format(fq_name_parts))
+
+        self._collection_name = '.'.join(fq_name_parts[1:3])
+
+        self._locate()
+
+    def _find_module(self, name_parts):
+        # synthesize empty inits for packages down through module_utils- we don't want to allow those to be shipped over, but the
+        # package hierarchy needs to exist
+        if len(name_parts) < 6:
+            self.source_code = ''
+            self.is_package = True
+            return True
+
+        # NB: we can't use pkgutil.get_data safely here, since we don't want to import/execute package/module code on
+        # the controller while analyzing/assembling the module, so we'll have to manually import the collection's
+        # Python package to locate it (import root collection, reassemble resource path beneath, fetch source)
+
+        collection_pkg_name = '.'.join(name_parts[0:3])
+        resource_base_path = os.path.join(*name_parts[3:])
+
+        src = None
+        # look for package_dir first, then module
+        try:
+            src = pkgutil.get_data(collection_pkg_name, to_native(os.path.join(resource_base_path, '__init__.py')))
+        except ImportError:
+            pass
+
+        # TODO: we might want to synthesize fake inits for py3-style packages, for now they're required beneath module_utils
+
+        if src is not None:  # empty string is OK
+            self.is_package = True
+        else:
+            try:
+                src = pkgutil.get_data(collection_pkg_name, to_native(resource_base_path + '.py'))
+            except ImportError:
+                pass
+
+        if src is None:  # empty string is OK
+            return False
+
+        self.source_code = src
+        return True
+
+    def _get_module_utils_remainder_parts(self, name_parts):
+        return name_parts[5:]  # eg, foo.bar for ansible_collections.ns.coll.plugins.module_utils.foo.bar
+
+
+def recursive_finder(name, module_fqn, module_data, zf):
+    """
+    Using ModuleDepFinder, make sure we have all of the module_utils files that
+    the module and its module_utils files needs. (no longer actually recursive)
+    :arg name: Name of the python module we're examining
+    :arg module_fqn: Fully qualified name of the python module we're scanning
+    :arg module_data: string Python code of the module we're scanning
+    :arg zf: An open :python:class:`zipfile.ZipFile` object that holds the Ansible module payload
+        which we're assembling
+    """
+
+    # py_module_cache maps python module names to a tuple of the code in the module
+    # and the pathname to the module.
+    # Here we pre-load it with modules which we create without bothering to
+    # read from actual files (In some cases, these need to differ from what ansible
+    # ships because they're namespace packages in the module)
+    # FIXME: do we actually want ns pkg behavior for these? Seems like they should just be forced to emptyish pkg stubs
+    py_module_cache = {
+        ('ansible',): (
+            b'from pkgutil import extend_path\n'
+            b'__path__=extend_path(__path__,__name__)\n'
+            b'__version__="' + to_bytes(__version__) +
+            b'"\n__author__="' + to_bytes(__author__) + b'"\n',
+            'ansible/__init__.py'),
+        ('ansible', 'module_utils'): (
+            b'from pkgutil import extend_path\n'
+            b'__path__=extend_path(__path__,__name__)\n',
+            'ansible/module_utils/__init__.py')}
+
+    module_utils_paths = [p for p in module_utils_loader._get_paths(subdirs=False) if os.path.isdir(p)]
+    module_utils_paths.append(_MODULE_UTILS_PATH)
+
+    # Parse the module code and find the imports of ansible.module_utils
+    try:
+        tree = compile(module_data, '<unknown>', 'exec', ast.PyCF_ONLY_AST)
+    except (SyntaxError, IndentationError) as e:
+        raise AnsibleError("Unable to import %s due to %s" % (name, e.msg))
+
+    finder = ModuleDepFinder(module_fqn, tree)
+
+    # the format of this set is a tuple of the module name and whether or not the import is ambiguous as a module name
+    # or an attribute of a module (eg from x.y import z <-- is z a module or an attribute of x.y?)
+    modules_to_process = [ModuleUtilsProcessEntry(m, True, False, is_optional=m in finder.optional_imports) for m in finder.submodules]
+
+    # HACK: basic is currently always required since module global init is currently tied up with AnsiballZ arg input
+    modules_to_process.append(ModuleUtilsProcessEntry(('ansible', 'module_utils', 'basic'), False, False, is_optional=False))
+
+    # we'll be adding new modules inline as we discover them, so just keep going til we've processed them all
+    while modules_to_process:
+        modules_to_process.sort()  # not strictly necessary, but nice to process things in predictable and repeatable order
+        py_module_name, is_ambiguous, child_is_redirected, is_optional = modules_to_process.pop(0)
+
+        if py_module_name in py_module_cache:
+            # this is normal; we'll often see the same module imported many times, but we only need to process it once
+            continue
+
+        if py_module_name[0:2] == ('ansible', 'module_utils'):
+            module_info = LegacyModuleUtilLocator(py_module_name, is_ambiguous=is_ambiguous,
+                                                  mu_paths=module_utils_paths, child_is_redirected=child_is_redirected)
+        elif py_module_name[0] == 'ansible_collections':
+            module_info = CollectionModuleUtilLocator(py_module_name, is_ambiguous=is_ambiguous,
+                                                      child_is_redirected=child_is_redirected, is_optional=is_optional)
+        else:
+            # FIXME: dot-joined result
+            display.warning('ModuleDepFinder improperly found a non-module_utils import %s'
+                            % [py_module_name])
+            continue
+
+        # Could not find the module.  Construct a helpful error message.
+        if not module_info.found:
+            if is_optional:
+                # this was a best-effort optional import that we couldn't find, oh well, move along...
+                continue
+            # FIXME: use dot-joined candidate names
+            msg = 'Could not find imported module support code for {0}.  Looked for ({1})'.format(module_fqn, module_info.candidate_names_joined)
+            raise AnsibleError(msg)
+
+        # check the cache one more time with the module we actually found, since the name could be different than the input
+        # eg, imported name vs module
+        if module_info.fq_name_parts in py_module_cache:
+            continue
+
+        # compile the source, process all relevant imported modules
+        try:
+            tree = compile(module_info.source_code, '<unknown>', 'exec', ast.PyCF_ONLY_AST)
+        except (SyntaxError, IndentationError) as e:
+            raise AnsibleError("Unable to import %s due to %s" % (module_info.fq_name_parts, e.msg))
+
+        finder = ModuleDepFinder('.'.join(module_info.fq_name_parts), tree, module_info.is_package)
+        modules_to_process.extend(ModuleUtilsProcessEntry(m, True, False, is_optional=m in finder.optional_imports)
+                                  for m in finder.submodules if m not in py_module_cache)
+
+        # we've processed this item, add it to the output list
+        py_module_cache[module_info.fq_name_parts] = (module_info.source_code, module_info.output_path)
+
+        # ensure we process all ancestor package inits
+        accumulated_pkg_name = []
+        for pkg in module_info.fq_name_parts[:-1]:
+            accumulated_pkg_name.append(pkg)  # we're accumulating this across iterations
+            normalized_name = tuple(accumulated_pkg_name)  # extra machinations to get a hashable type (list is not)
+            if normalized_name not in py_module_cache:
+                modules_to_process.append(ModuleUtilsProcessEntry(normalized_name, False, module_info.redirected, is_optional=is_optional))
+
+    for py_module_name in py_module_cache:
+        py_module_file_name = py_module_cache[py_module_name][1]
+
+        zf.writestr(py_module_file_name, py_module_cache[py_module_name][0])
+        mu_file = to_text(py_module_file_name, errors='surrogate_or_strict')
+        display.vvvvv("Including module_utils file %s" % mu_file)
+
+
+def _is_binary(b_module_data):
+    textchars = bytearray(set([7, 8, 9, 10, 12, 13, 27]) | set(range(0x20, 0x100)) - set([0x7f]))
+    start = b_module_data[:1024]
+    return bool(start.translate(None, textchars))
+
+
+def _get_ansible_module_fqn(module_path):
+    """
+    Get the fully qualified name for an ansible module based on its pathname
+
+    remote_module_fqn is the fully qualified name.  Like ansible.modules.system.ping
+    Or ansible_collections.Namespace.Collection_name.plugins.modules.ping
+    .. warning:: This function is for ansible modules only.  It won't work for other things
+        (non-module plugins, etc)
+    """
+    remote_module_fqn = None
+
+    # Is this a core module?
+    match = CORE_LIBRARY_PATH_RE.search(module_path)
+    if not match:
+        # Is this a module in a collection?
+        match = COLLECTION_PATH_RE.search(module_path)
+
+    # We can tell the FQN for core modules and collection modules
+    if match:
+        path = match.group('path')
+        if '.' in path:
+            # FQNs must be valid as python identifiers.  This sanity check has failed.
+            # we could check other things as well
+            raise ValueError('Module name (or path) was not a valid python identifier')
+
+        remote_module_fqn = '.'.join(path.split('/'))
+    else:
+        # Currently we do not handle modules in roles so we can end up here for that reason
+        raise ValueError("Unable to determine module's fully qualified name")
+
+    return remote_module_fqn
+
+
+def _add_module_to_zip(zf, remote_module_fqn, b_module_data):
+    """Add a module from ansible or from an ansible collection into the module zip"""
+    module_path_parts = remote_module_fqn.split('.')
+
+    # Write the module
+    module_path = '/'.join(module_path_parts) + '.py'
+    zf.writestr(module_path, b_module_data)
+
+    # Write the __init__.py's necessary to get there
+    if module_path_parts[0] == 'ansible':
+        # The ansible namespace is setup as part of the module_utils setup...
+        start = 2
+        existing_paths = frozenset()
+    else:
+        # ... but ansible_collections and other toplevels are not
+        start = 1
+        existing_paths = frozenset(zf.namelist())
+
+    for idx in range(start, len(module_path_parts)):
+        package_path = '/'.join(module_path_parts[:idx]) + '/__init__.py'
+        # If a collections module uses module_utils from a collection then most packages will have already been added by recursive_finder.
+        if package_path in existing_paths:
+            continue
+        # Note: We don't want to include more than one ansible module in a payload at this time
+        # so no need to fill the __init__.py with namespace code
+        zf.writestr(package_path, b'')
+
+
+def _find_module_utils(module_name, b_module_data, module_path, module_args, task_vars, templar, module_compression, async_timeout, become,
+                       become_method, become_user, become_password, become_flags, environment, remote_is_local=False):
+    """
+    Given the source of the module, convert it to a Jinja2 template to insert
+    module code and return whether it's a new or old style module.
+    """
+    module_substyle = module_style = 'old'
+
+    # module_style is something important to calling code (ActionBase).  It
+    # determines how arguments are formatted (json vs k=v) and whether
+    # a separate arguments file needs to be sent over the wire.
+    # module_substyle is extra information that's useful internally.  It tells
+    # us what we have to look to substitute in the module files and whether
+    # we're using module replacer or ansiballz to format the module itself.
+    if _is_binary(b_module_data):
+        module_substyle = module_style = 'binary'
+    elif REPLACER in b_module_data:
+        # Do REPLACER before from ansible.module_utils because we need make sure
+        # we substitute "from ansible.module_utils basic" for REPLACER
+        module_style = 'new'
+        module_substyle = 'python'
+        b_module_data = b_module_data.replace(REPLACER, b'from ansible.module_utils.basic import *')
+    elif NEW_STYLE_PYTHON_MODULE_RE.search(b_module_data):
+        module_style = 'new'
+        module_substyle = 'python'
+    elif REPLACER_WINDOWS in b_module_data:
+        module_style = 'new'
+        module_substyle = 'powershell'
+        b_module_data = b_module_data.replace(REPLACER_WINDOWS, b'#Requires -Module Ansible.ModuleUtils.Legacy')
+    elif re.search(b'#Requires -Module', b_module_data, re.IGNORECASE) \
+            or re.search(b'#Requires -Version', b_module_data, re.IGNORECASE)\
+            or re.search(b'#AnsibleRequires -OSVersion', b_module_data, re.IGNORECASE) \
+            or re.search(b'#AnsibleRequires -Powershell', b_module_data, re.IGNORECASE) \
+            or re.search(b'#AnsibleRequires -CSharpUtil', b_module_data, re.IGNORECASE):
+        module_style = 'new'
+        module_substyle = 'powershell'
+    elif REPLACER_JSONARGS in b_module_data:
+        module_style = 'new'
+        module_substyle = 'jsonargs'
+    elif b'WANT_JSON' in b_module_data:
+        module_substyle = module_style = 'non_native_want_json'
+
+    shebang = None
+    # Neither old-style, non_native_want_json nor binary modules should be modified
+    # except for the shebang line (Done by modify_module)
+    if module_style in ('old', 'non_native_want_json', 'binary'):
+        return b_module_data, module_style, shebang
+
+    output = BytesIO()
+
+    try:
+        remote_module_fqn = _get_ansible_module_fqn(module_path)
+    except ValueError:
+        # Modules in roles currently are not found by the fqn heuristic so we
+        # fallback to this.  This means that relative imports inside a module from
+        # a role may fail.  Absolute imports should be used for future-proofness.
+        # People should start writing collections instead of modules in roles so we
+        # may never fix this
+        display.debug('ANSIBALLZ: Could not determine module FQN')
+        remote_module_fqn = 'ansible.modules.%s' % module_name
+
+    if module_substyle == 'python':
+        params = dict(ANSIBLE_MODULE_ARGS=module_args,)
+        try:
+            python_repred_params = repr(json.dumps(params, cls=AnsibleJSONEncoder, vault_to_text=True))
+        except TypeError as e:
+            raise AnsibleError("Unable to pass options to module, they must be JSON serializable: %s" % to_native(e))
+
+        try:
+            compression_method = getattr(zipfile, module_compression)
+        except AttributeError:
+            display.warning(u'Bad module compression string specified: %s.  Using ZIP_STORED (no compression)' % module_compression)
+            compression_method = zipfile.ZIP_STORED
+
+        lookup_path = os.path.join(C.DEFAULT_LOCAL_TMP, 'ansiballz_cache')
+        cached_module_filename = os.path.join(lookup_path, "%s-%s" % (remote_module_fqn, module_compression))
+
+        zipdata = None
+        # Optimization -- don't lock if the module has already been cached
+        if os.path.exists(cached_module_filename):
+            display.debug('ANSIBALLZ: using cached module: %s' % cached_module_filename)
+            with open(cached_module_filename, 'rb') as module_data:
+                zipdata = module_data.read()
+        else:
+            if module_name in action_write_locks.action_write_locks:
+                display.debug('ANSIBALLZ: Using lock for %s' % module_name)
+                lock = action_write_locks.action_write_locks[module_name]
+            else:
+                # If the action plugin directly invokes the module (instead of
+                # going through a strategy) then we don't have a cross-process
+                # Lock specifically for this module.  Use the "unexpected
+                # module" lock instead
+                display.debug('ANSIBALLZ: Using generic lock for %s' % module_name)
+                lock = action_write_locks.action_write_locks[None]
+
+            display.debug('ANSIBALLZ: Acquiring lock')
+            with lock:
+                display.debug('ANSIBALLZ: Lock acquired: %s' % id(lock))
+                # Check that no other process has created this while we were
+                # waiting for the lock
+                if not os.path.exists(cached_module_filename):
+                    display.debug('ANSIBALLZ: Creating module')
+                    # Create the module zip data
+                    zipoutput = BytesIO()
+                    zf = zipfile.ZipFile(zipoutput, mode='w', compression=compression_method)
+
+                    # walk the module imports, looking for module_utils to send- they'll be added to the zipfile
+                    recursive_finder(module_name, remote_module_fqn, b_module_data, zf)
+
+                    display.debug('ANSIBALLZ: Writing module into payload')
+                    _add_module_to_zip(zf, remote_module_fqn, b_module_data)
+
+                    zf.close()
+                    zipdata = base64.b64encode(zipoutput.getvalue())
+
+                    # Write the assembled module to a temp file (write to temp
+                    # so that no one looking for the file reads a partially
+                    # written file)
+                    #
+                    # FIXME: Once split controller/remote is merged, this can be simplified to
+                    #        os.makedirs(lookup_path, exist_ok=True)
+                    if not os.path.exists(lookup_path):
+                        try:
+                            # Note -- if we have a global function to setup, that would
+                            # be a better place to run this
+                            os.makedirs(lookup_path)
+                        except OSError:
+                            # Multiple processes tried to create the directory. If it still does not
+                            # exist, raise the original exception.
+                            if not os.path.exists(lookup_path):
+                                raise
+                    display.debug('ANSIBALLZ: Writing module')
+                    with open(cached_module_filename + '-part', 'wb') as f:
+                        f.write(zipdata)
+
+                    # Rename the file into its final position in the cache so
+                    # future users of this module can read it off the
+                    # filesystem instead of constructing from scratch.
+                    display.debug('ANSIBALLZ: Renaming module')
+                    os.rename(cached_module_filename + '-part', cached_module_filename)
+                    display.debug('ANSIBALLZ: Done creating module')
+
+            if zipdata is None:
+                display.debug('ANSIBALLZ: Reading module after lock')
+                # Another process wrote the file while we were waiting for
+                # the write lock.  Go ahead and read the data from disk
+                # instead of re-creating it.
+                try:
+                    with open(cached_module_filename, 'rb') as f:
+                        zipdata = f.read()
+                except IOError:
+                    raise AnsibleError('A different worker process failed to create module file. '
+                                       'Look at traceback for that process for debugging information.')
+        zipdata = to_text(zipdata, errors='surrogate_or_strict')
+
+        o_interpreter, o_args = _extract_interpreter(b_module_data)
+        if o_interpreter is None:
+            o_interpreter = u'/usr/bin/python'
+
+        shebang, interpreter = _get_shebang(o_interpreter, task_vars, templar, o_args, remote_is_local=remote_is_local)
+
+        # FUTURE: the module cache entry should be invalidated if we got this value from a host-dependent source
+        rlimit_nofile = C.config.get_config_value('PYTHON_MODULE_RLIMIT_NOFILE', variables=task_vars)
+
+        if not isinstance(rlimit_nofile, int):
+            rlimit_nofile = int(templar.template(rlimit_nofile))
+
+        if rlimit_nofile:
+            rlimit = ANSIBALLZ_RLIMIT_TEMPLATE % dict(
+                rlimit_nofile=rlimit_nofile,
+            )
+        else:
+            rlimit = ''
+
+        coverage_config = os.environ.get('_ANSIBLE_COVERAGE_CONFIG')
+
+        if coverage_config:
+            coverage_output = os.environ['_ANSIBLE_COVERAGE_OUTPUT']
+
+            if coverage_output:
+                # Enable code coverage analysis of the module.
+                # This feature is for internal testing and may change without notice.
+                coverage = ANSIBALLZ_COVERAGE_TEMPLATE % dict(
+                    coverage_config=coverage_config,
+                    coverage_output=coverage_output,
+                )
+            else:
+                # Verify coverage is available without importing it.
+                # This will detect when a module would fail with coverage enabled with minimal overhead.
+                coverage = ANSIBALLZ_COVERAGE_CHECK_TEMPLATE
+        else:
+            coverage = ''
+
+        now = datetime.datetime.utcnow()
+        output.write(to_bytes(ACTIVE_ANSIBALLZ_TEMPLATE % dict(
+            zipdata=zipdata,
+            ansible_module=module_name,
+            module_fqn=remote_module_fqn,
+            params=python_repred_params,
+            shebang=shebang,
+            coding=ENCODING_STRING,
+            year=now.year,
+            month=now.month,
+            day=now.day,
+            hour=now.hour,
+            minute=now.minute,
+            second=now.second,
+            coverage=coverage,
+            rlimit=rlimit,
+        )))
+        b_module_data = output.getvalue()
+
+    elif module_substyle == 'powershell':
+        # Powershell/winrm don't actually make use of shebang so we can
+        # safely set this here.  If we let the fallback code handle this
+        # it can fail in the presence of the UTF8 BOM commonly added by
+        # Windows text editors
+        shebang = u'#!powershell'
+        # create the common exec wrapper payload and set that as the module_data
+        # bytes
+        b_module_data = ps_manifest._create_powershell_wrapper(
+            b_module_data, module_path, module_args, environment,
+            async_timeout, become, become_method, become_user, become_password,
+            become_flags, module_substyle, task_vars, remote_module_fqn
+        )
+
+    elif module_substyle == 'jsonargs':
+        module_args_json = to_bytes(json.dumps(module_args, cls=AnsibleJSONEncoder, vault_to_text=True))
+
+        # these strings could be included in a third-party module but
+        # officially they were included in the 'basic' snippet for new-style
+        # python modules (which has been replaced with something else in
+        # ansiballz) If we remove them from jsonargs-style module replacer
+        # then we can remove them everywhere.
+        python_repred_args = to_bytes(repr(module_args_json))
+        b_module_data = b_module_data.replace(REPLACER_VERSION, to_bytes(repr(__version__)))
+        b_module_data = b_module_data.replace(REPLACER_COMPLEX, python_repred_args)
+        b_module_data = b_module_data.replace(REPLACER_SELINUX, to_bytes(','.join(C.DEFAULT_SELINUX_SPECIAL_FS)))
+
+        # The main event -- substitute the JSON args string into the module
+        b_module_data = b_module_data.replace(REPLACER_JSONARGS, module_args_json)
+
+        facility = b'syslog.' + to_bytes(task_vars.get('ansible_syslog_facility', C.DEFAULT_SYSLOG_FACILITY), errors='surrogate_or_strict')
+        b_module_data = b_module_data.replace(b'syslog.LOG_USER', facility)
+
+    return (b_module_data, module_style, shebang)
+
+
+def _extract_interpreter(b_module_data):
+    """
+    Used to extract shebang expression from binary module data and return a text
+    string with the shebang, or None if no shebang is detected.
+    """
+
+    interpreter = None
+    args = []
+    b_lines = b_module_data.split(b"\n", 1)
+    if b_lines[0].startswith(b"#!"):
+        b_shebang = b_lines[0].strip()
+
+        # shlex.split needs text on Python 3
+        cli_split = shlex.split(to_text(b_shebang[2:], errors='surrogate_or_strict'))
+
+        # convert args to text
+        cli_split = [to_text(a, errors='surrogate_or_strict') for a in cli_split]
+        interpreter = cli_split[0]
+        args = cli_split[1:]
+
+    return interpreter, args
+
+
+def modify_module(module_name, module_path, module_args, templar, task_vars=None, module_compression='ZIP_STORED', async_timeout=0, become=False,
+                  become_method=None, become_user=None, become_password=None, become_flags=None, environment=None, remote_is_local=False):
+    """
+    Used to insert chunks of code into modules before transfer rather than
+    doing regular python imports.  This allows for more efficient transfer in
+    a non-bootstrapping scenario by not moving extra files over the wire and
+    also takes care of embedding arguments in the transferred modules.
+
+    This version is done in such a way that local imports can still be
+    used in the module code, so IDEs don't have to be aware of what is going on.
+
+    Example:
+
+    from ansible.module_utils.basic import *
+
+       ... will result in the insertion of basic.py into the module
+       from the module_utils/ directory in the source tree.
+
+    For powershell, this code effectively no-ops, as the exec wrapper requires access to a number of
+    properties not available here.
+
+    """
+    task_vars = {} if task_vars is None else task_vars
+    environment = {} if environment is None else environment
+
+    with open(module_path, 'rb') as f:
+
+        # read in the module source
+        b_module_data = f.read()
+
+    (b_module_data, module_style, shebang) = _find_module_utils(module_name, b_module_data, module_path, module_args, task_vars, templar, module_compression,
+                                                                async_timeout=async_timeout, become=become, become_method=become_method,
+                                                                become_user=become_user, become_password=become_password, become_flags=become_flags,
+                                                                environment=environment, remote_is_local=remote_is_local)
+
+    if module_style == 'binary':
+        return (b_module_data, module_style, to_text(shebang, nonstring='passthru'))
+    elif shebang is None:
+        interpreter, args = _extract_interpreter(b_module_data)
+        # No interpreter/shebang, assume a binary module?
+        if interpreter is not None:
+
+            shebang, new_interpreter = _get_shebang(interpreter, task_vars, templar, args, remote_is_local=remote_is_local)
+
+            # update shebang
+            b_lines = b_module_data.split(b"\n", 1)
+
+            if interpreter != new_interpreter:
+                b_lines[0] = to_bytes(shebang, errors='surrogate_or_strict', nonstring='passthru')
+
+            if os.path.basename(interpreter).startswith(u'python'):
+                b_lines.insert(1, b_ENCODING_STRING)
+
+            b_module_data = b"\n".join(b_lines)
+
+    return (b_module_data, module_style, shebang)
+
+
+def get_action_args_with_defaults(action, args, defaults, templar, redirected_names=None, action_groups=None):
+    if redirected_names:
+        resolved_action_name = redirected_names[-1]
+    else:
+        resolved_action_name = action
+
+    if redirected_names is not None:
+        msg = (
+            "Finding module_defaults for the action %s. "
+            "The caller passed a list of redirected action names, which is deprecated. "
+            "The task's resolved action should be provided as the first argument instead."
+        )
+        display.deprecated(msg % resolved_action_name, version='2.16')
+
+    # Get the list of groups that contain this action
+    if action_groups is None:
+        msg = (
+            "Finding module_defaults for action %s. "
+            "The caller has not passed the action_groups, so any "
+            "that may include this action will be ignored."
+        )
+        display.warning(msg=msg)
+        group_names = []
+    else:
+        group_names = action_groups.get(resolved_action_name, [])
+
+    tmp_args = {}
+    module_defaults = {}
+
+    # Merge latest defaults into dict, since they are a list of dicts
+    if isinstance(defaults, list):
+        for default in defaults:
+            module_defaults.update(default)
+
+    # module_defaults keys are static, but the values may be templated
+    module_defaults = templar.template(module_defaults)
+    for default in module_defaults:
+        if default.startswith('group/'):
+            group_name = default.split('group/')[-1]
+            if group_name in group_names:
+                tmp_args.update((module_defaults.get('group/%s' % group_name) or {}).copy())
+
+    # handle specific action defaults
+    tmp_args.update(module_defaults.get(resolved_action_name, {}).copy())
+
+    # direct args override all
+    tmp_args.update(args)
+
+    return tmp_args
diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
new file mode 100644
index 0000000..2449782
--- /dev/null
+++ b/lib/ansible/executor/play_iterator.py
@@ -0,0 +1,652 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import fnmatch
+
+from enum import IntEnum, IntFlag
+
+from ansible import constants as C
+from ansible.errors import AnsibleAssertionError
+from ansible.module_utils.parsing.convert_bool import boolean
+from ansible.playbook.block import Block
+from ansible.playbook.task import Task
+from ansible.utils.display import Display
+
+
+display = Display()
+
+
+__all__ = ['PlayIterator', 'IteratingStates', 'FailedStates']
+
+
+class IteratingStates(IntEnum):
+    SETUP = 0
+    TASKS = 1
+    RESCUE = 2
+    ALWAYS = 3
+    HANDLERS = 4
+    COMPLETE = 5
+
+
+class FailedStates(IntFlag):
+    NONE = 0
+    SETUP = 1
+    TASKS = 2
+    RESCUE = 4
+    ALWAYS = 8
+    HANDLERS = 16
+
+
+class HostState:
+    def __init__(self, blocks):
+        self._blocks = blocks[:]
+        self.handlers = []
+
+        self.cur_block = 0
+        self.cur_regular_task = 0
+        self.cur_rescue_task = 0
+        self.cur_always_task = 0
+        self.cur_handlers_task = 0
+        self.run_state = IteratingStates.SETUP
+        self.fail_state = FailedStates.NONE
+        self.pre_flushing_run_state = None
+        self.update_handlers = True
+        self.pending_setup = False
+        self.tasks_child_state = None
+        self.rescue_child_state = None
+        self.always_child_state = None
+        self.did_rescue = False
+        self.did_start_at_task = False
+
+    def __repr__(self):
+        return "HostState(%r)" % self._blocks
+
+    def __str__(self):
+        return ("HOST STATE: block=%d, task=%d, rescue=%d, always=%d, handlers=%d, run_state=%s, fail_state=%s, "
+                "pre_flushing_run_state=%s, update_handlers=%s, pending_setup=%s, "
+                "tasks child state? (%s), rescue child state? (%s), always child state? (%s), "
+                "did rescue? %s, did start at task? %s" % (
+                    self.cur_block,
+                    self.cur_regular_task,
+                    self.cur_rescue_task,
+                    self.cur_always_task,
+                    self.cur_handlers_task,
+                    self.run_state,
+                    self.fail_state,
+                    self.pre_flushing_run_state,
+                    self.update_handlers,
+                    self.pending_setup,
+                    self.tasks_child_state,
+                    self.rescue_child_state,
+                    self.always_child_state,
+                    self.did_rescue,
+                    self.did_start_at_task,
+                ))
+
+    def __eq__(self, other):
+        if not isinstance(other, HostState):
+            return False
+
+        for attr in ('_blocks',
+                     'cur_block', 'cur_regular_task', 'cur_rescue_task', 'cur_always_task', 'cur_handlers_task',
+                     'run_state', 'fail_state', 'pre_flushing_run_state', 'update_handlers', 'pending_setup',
+                     'tasks_child_state', 'rescue_child_state', 'always_child_state'):
+            if getattr(self, attr) != getattr(other, attr):
+                return False
+
+        return True
+
+    def get_current_block(self):
+        return self._blocks[self.cur_block]
+
+    def copy(self):
+        new_state = HostState(self._blocks)
+        new_state.handlers = self.handlers[:]
+        new_state.cur_block = self.cur_block
+        new_state.cur_regular_task = self.cur_regular_task
+        new_state.cur_rescue_task = self.cur_rescue_task
+        new_state.cur_always_task = self.cur_always_task
+        new_state.cur_handlers_task = self.cur_handlers_task
+        new_state.run_state = self.run_state
+        new_state.fail_state = self.fail_state
+        new_state.pre_flushing_run_state = self.pre_flushing_run_state
+        new_state.update_handlers = self.update_handlers
+        new_state.pending_setup = self.pending_setup
+        new_state.did_rescue = self.did_rescue
+        new_state.did_start_at_task = self.did_start_at_task
+        if self.tasks_child_state is not None:
+            new_state.tasks_child_state = self.tasks_child_state.copy()
+        if self.rescue_child_state is not None:
+            new_state.rescue_child_state = self.rescue_child_state.copy()
+        if self.always_child_state is not None:
+            new_state.always_child_state = self.always_child_state.copy()
+        return new_state
+
+
+class PlayIterator:
+
+    def __init__(self, inventory, play, play_context, variable_manager, all_vars, start_at_done=False):
+        self._play = play
+        self._blocks = []
+        self._variable_manager = variable_manager
+
+        setup_block = Block(play=self._play)
+        # Gathering facts with run_once would copy the facts from one host to
+        # the others.
+        setup_block.run_once = False
+        setup_task = Task(block=setup_block)
+        setup_task.action = 'gather_facts'
+        # TODO: hardcoded resolution here, but should use actual resolution code in the end,
+        #       in case of 'legacy' mismatch
+        setup_task.resolved_action = 'ansible.builtin.gather_facts'
+        setup_task.name = 'Gathering Facts'
+        setup_task.args = {}
+
+        # Unless play is specifically tagged, gathering should 'always' run
+        if not self._play.tags:
+            setup_task.tags = ['always']
+
+        # Default options to gather
+        for option in ('gather_subset', 'gather_timeout', 'fact_path'):
+            value = getattr(self._play, option, None)
+            if value is not None:
+                setup_task.args[option] = value
+
+        setup_task.set_loader(self._play._loader)
+        # short circuit fact gathering if the entire playbook is conditional
+        if self._play._included_conditional is not None:
+            setup_task.when = self._play._included_conditional[:]
+        setup_block.block = [setup_task]
+
+        setup_block = setup_block.filter_tagged_tasks(all_vars)
+        self._blocks.append(setup_block)
+
+        # keep flatten (no blocks) list of all tasks from the play
+        # used for the lockstep mechanism in the linear strategy
+        self.all_tasks = setup_block.get_tasks()
+
+        for block in self._play.compile():
+            new_block = block.filter_tagged_tasks(all_vars)
+            if new_block.has_tasks():
+                self._blocks.append(new_block)
+                self.all_tasks.extend(new_block.get_tasks())
+
+        # keep list of all handlers, it is copied into each HostState
+        # at the beginning of IteratingStates.HANDLERS
+        # the copy happens at each flush in order to restore the original
+        # list and remove any included handlers that might not be notified
+        # at the particular flush
+        self.handlers = [h for b in self._play.handlers for h in b.block]
+
+        self._host_states = {}
+        start_at_matched = False
+        batch = inventory.get_hosts(self._play.hosts, order=self._play.order)
+        self.batch_size = len(batch)
+        for host in batch:
+            self.set_state_for_host(host.name, HostState(blocks=self._blocks))
+            # if we're looking to start at a specific task, iterate through
+            # the tasks for this host until we find the specified task
+            if play_context.start_at_task is not None and not start_at_done:
+                while True:
+                    (s, task) = self.get_next_task_for_host(host, peek=True)
+                    if s.run_state == IteratingStates.COMPLETE:
+                        break
+                    if task.name == play_context.start_at_task or (task.name and fnmatch.fnmatch(task.name, play_context.start_at_task)) or \
+                       task.get_name() == play_context.start_at_task or fnmatch.fnmatch(task.get_name(), play_context.start_at_task):
+                        start_at_matched = True
+                        break
+                    self.set_state_for_host(host.name, s)
+
+                # finally, reset the host's state to IteratingStates.SETUP
+                if start_at_matched:
+                    self._host_states[host.name].did_start_at_task = True
+                    self._host_states[host.name].run_state = IteratingStates.SETUP
+
+        if start_at_matched:
+            # we have our match, so clear the start_at_task field on the
+            # play context to flag that we've started at a task (and future
+            # plays won't try to advance)
+            play_context.start_at_task = None
+
+        self.end_play = False
+        self.cur_task = 0
+
+    def get_host_state(self, host):
+        # Since we're using the PlayIterator to carry forward failed hosts,
+        # in the event that a previous host was not in the current inventory
+        # we create a stub state for it now
+        if host.name not in self._host_states:
+            self.set_state_for_host(host.name, HostState(blocks=[]))
+
+        return self._host_states[host.name].copy()
+
+    def cache_block_tasks(self, block):
+        display.deprecated(
+            'PlayIterator.cache_block_tasks is now noop due to the changes '
+            'in the way tasks are cached and is deprecated.',
+            version=2.16
+        )
+
+    def get_next_task_for_host(self, host, peek=False):
+
+        display.debug("getting the next task for host %s" % host.name)
+        s = self.get_host_state(host)
+
+        task = None
+        if s.run_state == IteratingStates.COMPLETE:
+            display.debug("host %s is done iterating, returning" % host.name)
+            return (s, None)
+
+        (s, task) = self._get_next_task_from_state(s, host=host)
+
+        if not peek:
+            self.set_state_for_host(host.name, s)
+
+        display.debug("done getting next task for host %s" % host.name)
+        display.debug(" ^ task is: %s" % task)
+        display.debug(" ^ state is: %s" % s)
+        return (s, task)
+
+    def _get_next_task_from_state(self, state, host):
+
+        task = None
+
+        # try and find the next task, given the current state.
+        while True:
+            # try to get the current block from the list of blocks, and
+            # if we run past the end of the list we know we're done with
+            # this block
+            try:
+                block = state._blocks[state.cur_block]
+            except IndexError:
+                state.run_state = IteratingStates.COMPLETE
+                return (state, None)
+
+            if state.run_state == IteratingStates.SETUP:
+                # First, we check to see if we were pending setup. If not, this is
+                # the first trip through IteratingStates.SETUP, so we set the pending_setup
+                # flag and try to determine if we do in fact want to gather facts for
+                # the specified host.
+                if not state.pending_setup:
+                    state.pending_setup = True
+
+                    # Gather facts if the default is 'smart' and we have not yet
+                    # done it for this host; or if 'explicit' and the play sets
+                    # gather_facts to True; or if 'implicit' and the play does
+                    # NOT explicitly set gather_facts to False.
+
+                    gathering = C.DEFAULT_GATHERING
+                    implied = self._play.gather_facts is None or boolean(self._play.gather_facts, strict=False)
+
+                    if (gathering == 'implicit' and implied) or \
+                       (gathering == 'explicit' and boolean(self._play.gather_facts, strict=False)) or \
+                       (gathering == 'smart' and implied and not (self._variable_manager._fact_cache.get(host.name, {}).get('_ansible_facts_gathered', False))):
+                        # The setup block is always self._blocks[0], as we inject it
+                        # during the play compilation in __init__ above.
+                        setup_block = self._blocks[0]
+                        if setup_block.has_tasks() and len(setup_block.block) > 0:
+                            task = setup_block.block[0]
+                else:
+                    # This is the second trip through IteratingStates.SETUP, so we clear
+                    # the flag and move onto the next block in the list while setting
+                    # the run state to IteratingStates.TASKS
+                    state.pending_setup = False
+
+                    state.run_state = IteratingStates.TASKS
+                    if not state.did_start_at_task:
+                        state.cur_block += 1
+                        state.cur_regular_task = 0
+                        state.cur_rescue_task = 0
+                        state.cur_always_task = 0
+                        state.tasks_child_state = None
+                        state.rescue_child_state = None
+                        state.always_child_state = None
+
+            elif state.run_state == IteratingStates.TASKS:
+                # clear the pending setup flag, since we're past that and it didn't fail
+                if state.pending_setup:
+                    state.pending_setup = False
+
+                # First, we check for a child task state that is not failed, and if we
+                # have one recurse into it for the next task. If we're done with the child
+                # state, we clear it and drop back to getting the next task from the list.
+                if state.tasks_child_state:
+                    (state.tasks_child_state, task) = self._get_next_task_from_state(state.tasks_child_state, host=host)
+                    if self._check_failed_state(state.tasks_child_state):
+                        # failed child state, so clear it and move into the rescue portion
+                        state.tasks_child_state = None
+                        self._set_failed_state(state)
+                    else:
+                        # get the next task recursively
+                        if task is None or state.tasks_child_state.run_state == IteratingStates.COMPLETE:
+                            # we're done with the child state, so clear it and continue
+                            # back to the top of the loop to get the next task
+                            state.tasks_child_state = None
+                            continue
+                else:
+                    # First here, we check to see if we've failed anywhere down the chain
+                    # of states we have, and if so we move onto the rescue portion. Otherwise,
+                    # we check to see if we've moved past the end of the list of tasks. If so,
+                    # we move into the always portion of the block, otherwise we get the next
+                    # task from the list.
+                    if self._check_failed_state(state):
+                        state.run_state = IteratingStates.RESCUE
+                    elif state.cur_regular_task >= len(block.block):
+                        state.run_state = IteratingStates.ALWAYS
+                    else:
+                        task = block.block[state.cur_regular_task]
+                        # if the current task is actually a child block, create a child
+                        # state for us to recurse into on the next pass
+                        if isinstance(task, Block):
+                            state.tasks_child_state = HostState(blocks=[task])
+                            state.tasks_child_state.run_state = IteratingStates.TASKS
+                            # since we've created the child state, clear the task
+                            # so we can pick up the child state on the next pass
+                            task = None
+                        state.cur_regular_task += 1
+
+            elif state.run_state == IteratingStates.RESCUE:
+                # The process here is identical to IteratingStates.TASKS, except instead
+                # we move into the always portion of the block.
+                if state.rescue_child_state:
+                    (state.rescue_child_state, task) = self._get_next_task_from_state(state.rescue_child_state, host=host)
+                    if self._check_failed_state(state.rescue_child_state):
+                        state.rescue_child_state = None
+                        self._set_failed_state(state)
+                    else:
+                        if task is None or state.rescue_child_state.run_state == IteratingStates.COMPLETE:
+                            state.rescue_child_state = None
+                            continue
+                else:
+                    if state.fail_state & FailedStates.RESCUE == FailedStates.RESCUE:
+                        state.run_state = IteratingStates.ALWAYS
+                    elif state.cur_rescue_task >= len(block.rescue):
+                        if len(block.rescue) > 0:
+                            state.fail_state = FailedStates.NONE
+                        state.run_state = IteratingStates.ALWAYS
+                        state.did_rescue = True
+                    else:
+                        task = block.rescue[state.cur_rescue_task]
+                        if isinstance(task, Block):
+                            state.rescue_child_state = HostState(blocks=[task])
+                            state.rescue_child_state.run_state = IteratingStates.TASKS
+                            task = None
+                        state.cur_rescue_task += 1
+
+            elif state.run_state == IteratingStates.ALWAYS:
+                # And again, the process here is identical to IteratingStates.TASKS, except
+                # instead we either move onto the next block in the list, or we set the
+                # run state to IteratingStates.COMPLETE in the event of any errors, or when we
+                # have hit the end of the list of blocks.
+                if state.always_child_state:
+                    (state.always_child_state, task) = self._get_next_task_from_state(state.always_child_state, host=host)
+                    if self._check_failed_state(state.always_child_state):
+                        state.always_child_state = None
+                        self._set_failed_state(state)
+                    else:
+                        if task is None or state.always_child_state.run_state == IteratingStates.COMPLETE:
+                            state.always_child_state = None
+                            continue
+                else:
+                    if state.cur_always_task >= len(block.always):
+                        if state.fail_state != FailedStates.NONE:
+                            state.run_state = IteratingStates.COMPLETE
+                        else:
+                            state.cur_block += 1
+                            state.cur_regular_task = 0
+                            state.cur_rescue_task = 0
+                            state.cur_always_task = 0
+                            state.run_state = IteratingStates.TASKS
+                            state.tasks_child_state = None
+                            state.rescue_child_state = None
+                            state.always_child_state = None
+                            state.did_rescue = False
+                    else:
+                        task = block.always[state.cur_always_task]
+                        if isinstance(task, Block):
+                            state.always_child_state = HostState(blocks=[task])
+                            state.always_child_state.run_state = IteratingStates.TASKS
+                            task = None
+                        state.cur_always_task += 1
+
+            elif state.run_state == IteratingStates.HANDLERS:
+                if state.update_handlers:
+                    # reset handlers for HostState since handlers from include_tasks
+                    # might be there from previous flush
+                    state.handlers = self.handlers[:]
+                    state.update_handlers = False
+                    state.cur_handlers_task = 0
+
+                if state.fail_state & FailedStates.HANDLERS == FailedStates.HANDLERS:
+                    state.update_handlers = True
+                    state.run_state = IteratingStates.COMPLETE
+                else:
+                    while True:
+                        try:
+                            task = state.handlers[state.cur_handlers_task]
+                        except IndexError:
+                            task = None
+                            state.run_state = state.pre_flushing_run_state
+                            state.update_handlers = True
+                            break
+                        else:
+                            state.cur_handlers_task += 1
+                            if task.is_host_notified(host):
+                                break
+
+            elif state.run_state == IteratingStates.COMPLETE:
+                return (state, None)
+
+            # if something above set the task, break out of the loop now
+            if task:
+                break
+
+        return (state, task)
+
+    def _set_failed_state(self, state):
+        if state.run_state == IteratingStates.SETUP:
+            state.fail_state |= FailedStates.SETUP
+            state.run_state = IteratingStates.COMPLETE
+        elif state.run_state == IteratingStates.TASKS:
+            if state.tasks_child_state is not None:
+                state.tasks_child_state = self._set_failed_state(state.tasks_child_state)
+            else:
+                state.fail_state |= FailedStates.TASKS
+                if state._blocks[state.cur_block].rescue:
+                    state.run_state = IteratingStates.RESCUE
+                elif state._blocks[state.cur_block].always:
+                    state.run_state = IteratingStates.ALWAYS
+                else:
+                    state.run_state = IteratingStates.COMPLETE
+        elif state.run_state == IteratingStates.RESCUE:
+            if state.rescue_child_state is not None:
+                state.rescue_child_state = self._set_failed_state(state.rescue_child_state)
+            else:
+                state.fail_state |= FailedStates.RESCUE
+                if state._blocks[state.cur_block].always:
+                    state.run_state = IteratingStates.ALWAYS
+                else:
+                    state.run_state = IteratingStates.COMPLETE
+        elif state.run_state == IteratingStates.ALWAYS:
+            if state.always_child_state is not None:
+                state.always_child_state = self._set_failed_state(state.always_child_state)
+            else:
+                state.fail_state |= FailedStates.ALWAYS
+                state.run_state = IteratingStates.COMPLETE
+        elif state.run_state == IteratingStates.HANDLERS:
+            state.fail_state |= FailedStates.HANDLERS
+            state.update_handlers = True
+            if state._blocks[state.cur_block].rescue:
+                state.run_state = IteratingStates.RESCUE
+            elif state._blocks[state.cur_block].always:
+                state.run_state = IteratingStates.ALWAYS
+            else:
+                state.run_state = IteratingStates.COMPLETE
+        return state
+
+    def mark_host_failed(self, host):
+        s = self.get_host_state(host)
+        display.debug("marking host %s failed, current state: %s" % (host, s))
+        s = self._set_failed_state(s)
+        display.debug("^ failed state is now: %s" % s)
+        self.set_state_for_host(host.name, s)
+        self._play._removed_hosts.append(host.name)
+
+    def get_failed_hosts(self):
+        return dict((host, True) for (host, state) in self._host_states.items() if self._check_failed_state(state))
+
+    def _check_failed_state(self, state):
+        if state is None:
+            return False
+        elif state.run_state == IteratingStates.RESCUE and self._check_failed_state(state.rescue_child_state):
+            return True
+        elif state.run_state == IteratingStates.ALWAYS and self._check_failed_state(state.always_child_state):
+            return True
+        elif state.run_state == IteratingStates.HANDLERS and state.fail_state & FailedStates.HANDLERS == FailedStates.HANDLERS:
+            return True
+        elif state.fail_state != FailedStates.NONE:
+            if state.run_state == IteratingStates.RESCUE and state.fail_state & FailedStates.RESCUE == 0:
+                return False
+            elif state.run_state == IteratingStates.ALWAYS and state.fail_state & FailedStates.ALWAYS == 0:
+                return False
+            else:
+                return not (state.did_rescue and state.fail_state & FailedStates.ALWAYS == 0)
+        elif state.run_state == IteratingStates.TASKS and self._check_failed_state(state.tasks_child_state):
+            cur_block = state._blocks[state.cur_block]
+            if len(cur_block.rescue) > 0 and state.fail_state & FailedStates.RESCUE == 0:
+                return False
+            else:
+                return True
+        return False
+
+    def is_failed(self, host):
+        s = self.get_host_state(host)
+        return self._check_failed_state(s)
+
+    def clear_host_errors(self, host):
+        self._clear_state_errors(self.get_state_for_host(host.name))
+
+    def _clear_state_errors(self, state: HostState) -> None:
+        state.fail_state = FailedStates.NONE
+
+        if state.tasks_child_state is not None:
+            self._clear_state_errors(state.tasks_child_state)
+        elif state.rescue_child_state is not None:
+            self._clear_state_errors(state.rescue_child_state)
+        elif state.always_child_state is not None:
+            self._clear_state_errors(state.always_child_state)
+
+    def get_active_state(self, state):
+        '''
+        Finds the active state, recursively if necessary when there are child states.
+        '''
+        if state.run_state == IteratingStates.TASKS and state.tasks_child_state is not None:
+            return self.get_active_state(state.tasks_child_state)
+        elif state.run_state == IteratingStates.RESCUE and state.rescue_child_state is not None:
+            return self.get_active_state(state.rescue_child_state)
+        elif state.run_state == IteratingStates.ALWAYS and state.always_child_state is not None:
+            return self.get_active_state(state.always_child_state)
+        return state
+
+    def is_any_block_rescuing(self, state):
+        '''
+        Given the current HostState state, determines if the current block, or any child blocks,
+        are in rescue mode.
+        '''
+        if state.run_state == IteratingStates.TASKS and state.get_current_block().rescue:
+            return True
+        if state.tasks_child_state is not None:
+            return self.is_any_block_rescuing(state.tasks_child_state)
+        if state.rescue_child_state is not None:
+            return self.is_any_block_rescuing(state.rescue_child_state)
+        if state.always_child_state is not None:
+            return self.is_any_block_rescuing(state.always_child_state)
+        return False
+
+    def get_original_task(self, host, task):
+        display.deprecated(
+            'PlayIterator.get_original_task is now noop due to the changes '
+            'in the way tasks are cached and is deprecated.',
+            version=2.16
+        )
+        return (None, None)
+
+    def _insert_tasks_into_state(self, state, task_list):
+        # if we've failed at all, or if the task list is empty, just return the current state
+        if (state.fail_state != FailedStates.NONE and state.run_state == IteratingStates.TASKS) or not task_list:
+            return state
+
+        if state.run_state == IteratingStates.TASKS:
+            if state.tasks_child_state:
+                state.tasks_child_state = self._insert_tasks_into_state(state.tasks_child_state, task_list)
+            else:
+                target_block = state._blocks[state.cur_block].copy()
+                before = target_block.block[:state.cur_regular_task]
+                after = target_block.block[state.cur_regular_task:]
+                target_block.block = before + task_list + after
+                state._blocks[state.cur_block] = target_block
+        elif state.run_state == IteratingStates.RESCUE:
+            if state.rescue_child_state:
+                state.rescue_child_state = self._insert_tasks_into_state(state.rescue_child_state, task_list)
+            else:
+                target_block = state._blocks[state.cur_block].copy()
+                before = target_block.rescue[:state.cur_rescue_task]
+                after = target_block.rescue[state.cur_rescue_task:]
+                target_block.rescue = before + task_list + after
+                state._blocks[state.cur_block] = target_block
+        elif state.run_state == IteratingStates.ALWAYS:
+            if state.always_child_state:
+                state.always_child_state = self._insert_tasks_into_state(state.always_child_state, task_list)
+            else:
+                target_block = state._blocks[state.cur_block].copy()
+                before = target_block.always[:state.cur_always_task]
+                after = target_block.always[state.cur_always_task:]
+                target_block.always = before + task_list + after
+                state._blocks[state.cur_block] = target_block
+        elif state.run_state == IteratingStates.HANDLERS:
+            state.handlers[state.cur_handlers_task:state.cur_handlers_task] = [h for b in task_list for h in b.block]
+
+        return state
+
+    def add_tasks(self, host, task_list):
+        self.set_state_for_host(host.name, self._insert_tasks_into_state(self.get_host_state(host), task_list))
+
+    @property
+    def host_states(self):
+        return self._host_states
+
+    def get_state_for_host(self, hostname: str) -> HostState:
+        return self._host_states[hostname]
+
+    def set_state_for_host(self, hostname: str, state: HostState) -> None:
+        if not isinstance(state, HostState):
+            raise AnsibleAssertionError('Expected state to be a HostState but was a %s' % type(state))
+        self._host_states[hostname] = state
+
+    def set_run_state_for_host(self, hostname: str, run_state: IteratingStates) -> None:
+        if not isinstance(run_state, IteratingStates):
+            raise AnsibleAssertionError('Expected run_state to be a IteratingStates but was %s' % (type(run_state)))
+        self._host_states[hostname].run_state = run_state
+
+    def set_fail_state_for_host(self, hostname: str, fail_state: FailedStates) -> None:
+        if not isinstance(fail_state, FailedStates):
+            raise AnsibleAssertionError('Expected fail_state to be a FailedStates but was %s' % (type(fail_state)))
+        self._host_states[hostname].fail_state = fail_state
diff --git a/lib/ansible/executor/playbook_executor.py b/lib/ansible/executor/playbook_executor.py
new file mode 100644
index 0000000..e8b2a3d
--- /dev/null
+++ b/lib/ansible/executor/playbook_executor.py
@@ -0,0 +1,335 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+
+from ansible import constants as C
+from ansible import context
+from ansible.executor.task_queue_manager import TaskQueueManager, AnsibleEndPlay
+from ansible.module_utils._text import to_text
+from ansible.module_utils.parsing.convert_bool import boolean
+from ansible.plugins.loader import become_loader, connection_loader, shell_loader
+from ansible.playbook import Playbook
+from ansible.template import Templar
+from ansible.utils.helpers import pct_to_int
+from ansible.utils.collection_loader import AnsibleCollectionConfig
+from ansible.utils.collection_loader._collection_finder import _get_collection_name_from_path, _get_collection_playbook_path
+from ansible.utils.path import makedirs_safe
+from ansible.utils.ssh_functions import set_default_transport
+from ansible.utils.display import Display
+
+
+display = Display()
+
+
+class PlaybookExecutor:
+
+    '''
+    This is the primary class for executing playbooks, and thus the
+    basis for bin/ansible-playbook operation.
+    '''
+
+    def __init__(self, playbooks, inventory, variable_manager, loader, passwords):
+        self._playbooks = playbooks
+        self._inventory = inventory
+        self._variable_manager = variable_manager
+        self._loader = loader
+        self.passwords = passwords
+        self._unreachable_hosts = dict()
+
+        if context.CLIARGS.get('listhosts') or context.CLIARGS.get('listtasks') or \
+                context.CLIARGS.get('listtags') or context.CLIARGS.get('syntax'):
+            self._tqm = None
+        else:
+            self._tqm = TaskQueueManager(
+                inventory=inventory,
+                variable_manager=variable_manager,
+                loader=loader,
+                passwords=self.passwords,
+                forks=context.CLIARGS.get('forks'),
+            )
+
+        # Note: We run this here to cache whether the default ansible ssh
+        # executable supports control persist.  Sometime in the future we may
+        # need to enhance this to check that ansible_ssh_executable specified
+        # in inventory is also cached.  We can't do this caching at the point
+        # where it is used (in task_executor) because that is post-fork and
+        # therefore would be discarded after every task.
+        set_default_transport()
+
+    def run(self):
+        '''
+        Run the given playbook, based on the settings in the play which
+        may limit the runs to serialized groups, etc.
+        '''
+
+        result = 0
+        entrylist = []
+        entry = {}
+        try:
+            # preload become/connection/shell to set config defs cached
+            list(connection_loader.all(class_only=True))
+            list(shell_loader.all(class_only=True))
+            list(become_loader.all(class_only=True))
+
+            for playbook in self._playbooks:
+
+                # deal with FQCN
+                resource = _get_collection_playbook_path(playbook)
+                if resource is not None:
+                    playbook_path = resource[1]
+                    playbook_collection = resource[2]
+                else:
+                    playbook_path = playbook
+                    # not fqcn, but might still be colleciotn playbook
+                    playbook_collection = _get_collection_name_from_path(playbook)
+
+                if playbook_collection:
+                    display.warning("running playbook inside collection {0}".format(playbook_collection))
+                    AnsibleCollectionConfig.default_collection = playbook_collection
+                else:
+                    AnsibleCollectionConfig.default_collection = None
+
+                pb = Playbook.load(playbook_path, variable_manager=self._variable_manager, loader=self._loader)
+                # FIXME: move out of inventory self._inventory.set_playbook_basedir(os.path.realpath(os.path.dirname(playbook_path)))
+
+                if self._tqm is None:  # we are doing a listing
+                    entry = {'playbook': playbook_path}
+                    entry['plays'] = []
+                else:
+                    # make sure the tqm has callbacks loaded
+                    self._tqm.load_callbacks()
+                    self._tqm.send_callback('v2_playbook_on_start', pb)
+
+                i = 1
+                plays = pb.get_plays()
+                display.vv(u'%d plays in %s' % (len(plays), to_text(playbook_path)))
+
+                for play in plays:
+                    if play._included_path is not None:
+                        self._loader.set_basedir(play._included_path)
+                    else:
+                        self._loader.set_basedir(pb._basedir)
+
+                    # clear any filters which may have been applied to the inventory
+                    self._inventory.remove_restriction()
+
+                    # Allow variables to be used in vars_prompt fields.
+                    all_vars = self._variable_manager.get_vars(play=play)
+                    templar = Templar(loader=self._loader, variables=all_vars)
+                    setattr(play, 'vars_prompt', templar.template(play.vars_prompt))
+
+                    # FIXME: this should be a play 'sub object' like loop_control
+                    if play.vars_prompt:
+                        for var in play.vars_prompt:
+                            vname = var['name']
+                            prompt = var.get("prompt", vname)
+                            default = var.get("default", None)
+                            private = boolean(var.get("private", True))
+                            confirm = boolean(var.get("confirm", False))
+                            encrypt = var.get("encrypt", None)
+                            salt_size = var.get("salt_size", None)
+                            salt = var.get("salt", None)
+                            unsafe = var.get("unsafe", None)
+
+                            if vname not in self._variable_manager.extra_vars:
+                                if self._tqm:
+                                    self._tqm.send_callback('v2_playbook_on_vars_prompt', vname, private, prompt, encrypt, confirm, salt_size, salt,
+                                                            default, unsafe)
+                                    play.vars[vname] = display.do_var_prompt(vname, private, prompt, encrypt, confirm, salt_size, salt, default, unsafe)
+                                else:  # we are either in --list-<option> or syntax check
+                                    play.vars[vname] = default
+
+                    # Post validate so any play level variables are templated
+                    all_vars = self._variable_manager.get_vars(play=play)
+                    templar = Templar(loader=self._loader, variables=all_vars)
+                    play.post_validate(templar)
+
+                    if context.CLIARGS['syntax']:
+                        continue
+
+                    if self._tqm is None:
+                        # we are just doing a listing
+                        entry['plays'].append(play)
+
+                    else:
+                        self._tqm._unreachable_hosts.update(self._unreachable_hosts)
+
+                        previously_failed = len(self._tqm._failed_hosts)
+                        previously_unreachable = len(self._tqm._unreachable_hosts)
+
+                        break_play = False
+                        # we are actually running plays
+                        batches = self._get_serialized_batches(play)
+                        if len(batches) == 0:
+                            self._tqm.send_callback('v2_playbook_on_play_start', play)
+                            self._tqm.send_callback('v2_playbook_on_no_hosts_matched')
+                        for batch in batches:
+                            # restrict the inventory to the hosts in the serialized batch
+                            self._inventory.restrict_to_hosts(batch)
+                            # and run it...
+                            try:
+                                result = self._tqm.run(play=play)
+                            except AnsibleEndPlay as e:
+                                result = e.result
+                                break
+
+                            # break the play if the result equals the special return code
+                            if result & self._tqm.RUN_FAILED_BREAK_PLAY != 0:
+                                result = self._tqm.RUN_FAILED_HOSTS
+                                break_play = True
+
+                            # check the number of failures here, to see if they're above the maximum
+                            # failure percentage allowed, or if any errors are fatal. If either of those
+                            # conditions are met, we break out, otherwise we only break out if the entire
+                            # batch failed
+                            failed_hosts_count = len(self._tqm._failed_hosts) + len(self._tqm._unreachable_hosts) - \
+                                (previously_failed + previously_unreachable)
+
+                            if len(batch) == failed_hosts_count:
+                                break_play = True
+                                break
+
+                            # update the previous counts so they don't accumulate incorrectly
+                            # over multiple serial batches
+                            previously_failed += len(self._tqm._failed_hosts) - previously_failed
+                            previously_unreachable += len(self._tqm._unreachable_hosts) - previously_unreachable
+
+                            # save the unreachable hosts from this batch
+                            self._unreachable_hosts.update(self._tqm._unreachable_hosts)
+
+                        if break_play:
+                            break
+
+                    i = i + 1  # per play
+
+                if entry:
+                    entrylist.append(entry)  # per playbook
+
+                # send the stats callback for this playbook
+                if self._tqm is not None:
+                    if C.RETRY_FILES_ENABLED:
+                        retries = set(self._tqm._failed_hosts.keys())
+                        retries.update(self._tqm._unreachable_hosts.keys())
+                        retries = sorted(retries)
+                        if len(retries) > 0:
+                            if C.RETRY_FILES_SAVE_PATH:
+                                basedir = C.RETRY_FILES_SAVE_PATH
+                            elif playbook_path:
+                                basedir = os.path.dirname(os.path.abspath(playbook_path))
+                            else:
+                                basedir = '~/'
+
+                            (retry_name, _) = os.path.splitext(os.path.basename(playbook_path))
+                            filename = os.path.join(basedir, "%s.retry" % retry_name)
+                            if self._generate_retry_inventory(filename, retries):
+                                display.display("\tto retry, use: --limit @%s\n" % filename)
+
+                    self._tqm.send_callback('v2_playbook_on_stats', self._tqm._stats)
+
+                # if the last result wasn't zero, break out of the playbook file name loop
+                if result != 0:
+                    break
+
+            if entrylist:
+                return entrylist
+
+        finally:
+            if self._tqm is not None:
+                self._tqm.cleanup()
+            if self._loader:
+                self._loader.cleanup_all_tmp_files()
+
+        if context.CLIARGS['syntax']:
+            display.display("No issues encountered")
+            return result
+
+        if context.CLIARGS['start_at_task'] and not self._tqm._start_at_done:
+            display.error(
+                "No matching task \"%s\" found."
+                " Note: --start-at-task can only follow static includes."
+                % context.CLIARGS['start_at_task']
+            )
+
+        return result
+
+    def _get_serialized_batches(self, play):
+        '''
+        Returns a list of hosts, subdivided into batches based on
+        the serial size specified in the play.
+        '''
+
+        # make sure we have a unique list of hosts
+        all_hosts = self._inventory.get_hosts(play.hosts, order=play.order)
+        all_hosts_len = len(all_hosts)
+
+        # the serial value can be listed as a scalar or a list of
+        # scalars, so we make sure it's a list here
+        serial_batch_list = play.serial
+        if len(serial_batch_list) == 0:
+            serial_batch_list = [-1]
+
+        cur_item = 0
+        serialized_batches = []
+
+        while len(all_hosts) > 0:
+            # get the serial value from current item in the list
+            serial = pct_to_int(serial_batch_list[cur_item], all_hosts_len)
+
+            # if the serial count was not specified or is invalid, default to
+            # a list of all hosts, otherwise grab a chunk of the hosts equal
+            # to the current serial item size
+            if serial <= 0:
+                serialized_batches.append(all_hosts)
+                break
+            else:
+                play_hosts = []
+                for x in range(serial):
+                    if len(all_hosts) > 0:
+                        play_hosts.append(all_hosts.pop(0))
+
+                serialized_batches.append(play_hosts)
+
+            # increment the current batch list item number, and if we've hit
+            # the end keep using the last element until we've consumed all of
+            # the hosts in the inventory
+            cur_item += 1
+            if cur_item > len(serial_batch_list) - 1:
+                cur_item = len(serial_batch_list) - 1
+
+        return serialized_batches
+
+    def _generate_retry_inventory(self, retry_path, replay_hosts):
+        '''
+        Called when a playbook run fails. It generates an inventory which allows
+        re-running on ONLY the failed hosts.  This may duplicate some variable
+        information in group_vars/host_vars but that is ok, and expected.
+        '''
+        try:
+            makedirs_safe(os.path.dirname(retry_path))
+            with open(retry_path, 'w') as fd:
+                for x in replay_hosts:
+                    fd.write("%s\n" % x)
+        except Exception as e:
+            display.warning("Could not create retry file '%s'.\n\t%s" % (retry_path, to_text(e)))
+            return False
+
+        return True
diff --git a/lib/ansible/executor/powershell/__init__.py b/lib/ansible/executor/powershell/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/lib/ansible/executor/powershell/__init__.py
diff --git a/lib/ansible/executor/powershell/async_watchdog.ps1 b/lib/ansible/executor/powershell/async_watchdog.ps1
new file mode 100644
index 0000000..c2138e3
--- /dev/null
+++ b/lib/ansible/executor/powershell/async_watchdog.ps1
@@ -0,0 +1,117 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+# help with debugging errors as we don't have visibility of this running process
+trap {
+    $watchdog_path = "$($env:TEMP)\ansible-async-watchdog-error-$(Get-Date -Format "yyyy-MM-ddTHH-mm-ss.ffffZ").txt"
+    $error_msg = "Error while running the async exec wrapper`r`n$(Format-AnsibleException -ErrorRecord $_)"
+    Set-Content -Path $watchdog_path -Value $error_msg
+    break
+}
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting async_watchdog" "async_watchdog"
+
+# pop 0th action as entrypoint
+$payload.actions = $payload.actions[1..99]
+
+$actions = $Payload.actions
+$entrypoint = $payload.($actions[0])
+$entrypoint = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($entrypoint))
+
+$resultfile_path = $payload.async_results_path
+$max_exec_time_sec = $payload.async_timeout_sec
+
+Write-AnsibleLog "INFO - deserializing existing result file args at: '$resultfile_path'" "async_watchdog"
+if (-not (Test-Path -Path $resultfile_path)) {
+    $msg = "result file at '$resultfile_path' does not exist"
+    Write-AnsibleLog "ERROR - $msg" "async_watchdog"
+    throw $msg
+}
+$result_json = Get-Content -Path $resultfile_path -Raw
+Write-AnsibleLog "INFO - result file json is: $result_json" "async_watchdog"
+$result = ConvertFrom-AnsibleJson -InputObject $result_json
+
+Write-AnsibleLog "INFO - creating async runspace" "async_watchdog"
+$rs = [RunspaceFactory]::CreateRunspace()
+$rs.Open()
+
+Write-AnsibleLog "INFO - creating async PowerShell pipeline" "async_watchdog"
+$ps = [PowerShell]::Create()
+$ps.Runspace = $rs
+
+# these functions are set in exec_wrapper
+Write-AnsibleLog "INFO - adding global functions to PowerShell pipeline script" "async_watchdog"
+$ps.AddScript($script:common_functions).AddStatement() > $null
+$ps.AddScript($script:wrapper_functions).AddStatement() > $null
+$function_params = @{
+    Name = "common_functions"
+    Value = $script:common_functions
+    Scope = "script"
+}
+$ps.AddCommand("Set-Variable").AddParameters($function_params).AddStatement() > $null
+
+Write-AnsibleLog "INFO - adding $($actions[0]) to PowerShell pipeline script" "async_watchdog"
+$ps.AddScript($entrypoint).AddArgument($payload) > $null
+
+Write-AnsibleLog "INFO - async job start, calling BeginInvoke()" "async_watchdog"
+$job_async_result = $ps.BeginInvoke()
+
+Write-AnsibleLog "INFO - waiting '$max_exec_time_sec' seconds for async job to complete" "async_watchdog"
+$job_async_result.AsyncWaitHandle.WaitOne($max_exec_time_sec * 1000) > $null
+$result.finished = 1
+
+if ($job_async_result.IsCompleted) {
+    Write-AnsibleLog "INFO - async job completed, calling EndInvoke()" "async_watchdog"
+
+    $job_output = $ps.EndInvoke($job_async_result)
+    $job_error = $ps.Streams.Error
+
+    Write-AnsibleLog "INFO - raw module stdout:`r`n$($job_output | Out-String)" "async_watchdog"
+    if ($job_error) {
+        Write-AnsibleLog "WARN - raw module stderr:`r`n$($job_error | Out-String)" "async_watchdog"
+    }
+
+    # write success/output/error to result object
+    # TODO: cleanse leading/trailing junk
+    try {
+        Write-AnsibleLog "INFO - deserializing Ansible stdout" "async_watchdog"
+        $module_result = ConvertFrom-AnsibleJson -InputObject $job_output
+        # TODO: check for conflicting keys
+        $result = $result + $module_result
+    }
+    catch {
+        $result.failed = $true
+        $result.msg = "failed to parse module output: $($_.Exception.Message)"
+        # return output back to Ansible to help with debugging errors
+        $result.stdout = $job_output | Out-String
+        $result.stderr = $job_error | Out-String
+    }
+
+    $result_json = ConvertTo-Json -InputObject $result -Depth 99 -Compress
+    Set-Content -Path $resultfile_path -Value $result_json
+
+    Write-AnsibleLog "INFO - wrote output to $resultfile_path" "async_watchdog"
+}
+else {
+    Write-AnsibleLog "ERROR - reached timeout on async job, stopping job" "async_watchdog"
+    $ps.BeginStop($null, $null)  > $null # best effort stop
+
+    # write timeout to result object
+    $result.failed = $true
+    $result.msg = "timed out waiting for module completion"
+    $result_json = ConvertTo-Json -InputObject $result -Depth 99 -Compress
+    Set-Content -Path $resultfile_path -Value $result_json
+
+    Write-AnsibleLog "INFO - wrote timeout to '$resultfile_path'" "async_watchdog"
+}
+
+# in the case of a hung pipeline, this will cause the process to stay alive until it's un-hung...
+#$rs.Close() | Out-Null
+
+Write-AnsibleLog "INFO - ending async_watchdog" "async_watchdog"
diff --git a/lib/ansible/executor/powershell/async_wrapper.ps1 b/lib/ansible/executor/powershell/async_wrapper.ps1
new file mode 100644
index 0000000..0cd640f
--- /dev/null
+++ b/lib/ansible/executor/powershell/async_wrapper.ps1
@@ -0,0 +1,174 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting async_wrapper" "async_wrapper"
+
+if (-not $Payload.environment.ContainsKey("ANSIBLE_ASYNC_DIR")) {
+    Write-AnsibleError -Message "internal error: the environment variable ANSIBLE_ASYNC_DIR is not set and is required for an async task"
+    $host.SetShouldExit(1)
+    return
+}
+$async_dir = [System.Environment]::ExpandEnvironmentVariables($Payload.environment.ANSIBLE_ASYNC_DIR)
+
+# calculate the result path so we can include it in the worker payload
+$jid = $Payload.async_jid
+$local_jid = $jid + "." + $pid
+
+$results_path = [System.IO.Path]::Combine($async_dir, $local_jid)
+
+Write-AnsibleLog "INFO - creating async results path at '$results_path'" "async_wrapper"
+
+$Payload.async_results_path = $results_path
+[System.IO.Directory]::CreateDirectory([System.IO.Path]::GetDirectoryName($results_path)) > $null
+
+# we use Win32_Process to escape the current process job, CreateProcess with a
+# breakaway flag won't work for psrp as the psrp process does not have breakaway
+# rights. Unfortunately we can't read/write to the spawned process as we can't
+# inherit the handles. We use a locked down named pipe to send the exec_wrapper
+# payload. Anonymous pipes won't work as the spawned process will not be a child
+# of the current one and will not be able to inherit the handles
+
+# pop the async_wrapper action so we don't get stuck in a loop and create new
+# exec_wrapper for our async process
+$Payload.actions = $Payload.actions[1..99]
+$payload_json = ConvertTo-Json -InputObject $Payload -Depth 99 -Compress
+
+#
+$exec_wrapper = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.exec_wrapper))
+$exec_wrapper += "`0`0`0`0" + $payload_json
+$payload_bytes = [System.Text.Encoding]::UTF8.GetBytes($exec_wrapper)
+$pipe_name = "ansible-async-$jid-$([guid]::NewGuid())"
+
+# template the async process command line with the payload details
+$bootstrap_wrapper = {
+    # help with debugging errors as we loose visibility of the process output
+    # from here on
+    trap {
+        $wrapper_path = "$($env:TEMP)\ansible-async-wrapper-error-$(Get-Date -Format "yyyy-MM-ddTHH-mm-ss.ffffZ").txt"
+        $error_msg = "Error while running the async exec wrapper`r`n$($_ | Out-String)`r`n$($_.ScriptStackTrace)"
+        Set-Content -Path $wrapper_path -Value $error_msg
+        break
+    }
+
+    &chcp.com 65001 > $null
+
+    # store the pipe name and no. of bytes to read, these are populated before
+    # before the process is created - do not remove or changed
+    $pipe_name = ""
+    $bytes_length = 0
+
+    $input_bytes = New-Object -TypeName byte[] -ArgumentList $bytes_length
+    $pipe = New-Object -TypeName System.IO.Pipes.NamedPipeClientStream -ArgumentList @(
+        ".", # localhost
+        $pipe_name,
+        [System.IO.Pipes.PipeDirection]::In,
+        [System.IO.Pipes.PipeOptions]::None,
+        [System.Security.Principal.TokenImpersonationLevel]::Anonymous
+    )
+    try {
+        $pipe.Connect()
+        $pipe.Read($input_bytes, 0, $bytes_length) > $null
+    }
+    finally {
+        $pipe.Close()
+    }
+    $exec = [System.Text.Encoding]::UTF8.GetString($input_bytes)
+    $exec_parts = $exec.Split(@("`0`0`0`0"), 2, [StringSplitOptions]::RemoveEmptyEntries)
+    Set-Variable -Name json_raw -Value $exec_parts[1]
+    $exec = [ScriptBlock]::Create($exec_parts[0])
+    &$exec
+}
+
+$bootstrap_wrapper = $bootstrap_wrapper.ToString().Replace('$pipe_name = ""', "`$pipe_name = `"$pipe_name`"")
+$bootstrap_wrapper = $bootstrap_wrapper.Replace('$bytes_length = 0', "`$bytes_length = $($payload_bytes.Count)")
+$encoded_command = [System.Convert]::ToBase64String([System.Text.Encoding]::Unicode.GetBytes($bootstrap_wrapper))
+$pwsh_path = "$env:SystemRoot\System32\WindowsPowerShell\v1.0\powershell.exe"
+$exec_args = "`"$pwsh_path`" -NonInteractive -NoProfile -ExecutionPolicy Bypass -EncodedCommand $encoded_command"
+
+# create a named pipe that is set to allow only the current user read access
+$current_user = ([Security.Principal.WindowsIdentity]::GetCurrent()).User
+$pipe_sec = New-Object -TypeName System.IO.Pipes.PipeSecurity
+$pipe_ar = New-Object -TypeName System.IO.Pipes.PipeAccessRule -ArgumentList @(
+    $current_user,
+    [System.IO.Pipes.PipeAccessRights]::Read,
+    [System.Security.AccessControl.AccessControlType]::Allow
+)
+$pipe_sec.AddAccessRule($pipe_ar)
+
+Write-AnsibleLog "INFO - creating named pipe '$pipe_name'" "async_wrapper"
+$pipe = New-Object -TypeName System.IO.Pipes.NamedPipeServerStream -ArgumentList @(
+    $pipe_name,
+    [System.IO.Pipes.PipeDirection]::Out,
+    1,
+    [System.IO.Pipes.PipeTransmissionMode]::Byte,
+    [System.IO.Pipes.PipeOptions]::Asynchronous,
+    0,
+    0,
+    $pipe_sec
+)
+
+try {
+    Write-AnsibleLog "INFO - creating async process '$exec_args'" "async_wrapper"
+    $process = Invoke-CimMethod -ClassName Win32_Process -Name Create -Arguments @{CommandLine = $exec_args }
+    $rc = $process.ReturnValue
+
+    Write-AnsibleLog "INFO - return value from async process exec: $rc" "async_wrapper"
+    if ($rc -ne 0) {
+        $error_msg = switch ($rc) {
+            2 { "Access denied" }
+            3 { "Insufficient privilege" }
+            8 { "Unknown failure" }
+            9 { "Path not found" }
+            21 { "Invalid parameter" }
+            default { "Other" }
+        }
+        throw "Failed to start async process: $rc ($error_msg)"
+    }
+    $watchdog_pid = $process.ProcessId
+    Write-AnsibleLog "INFO - created async process PID: $watchdog_pid" "async_wrapper"
+
+    # populate initial results before we send the async data to avoid result race
+    $result = @{
+        started = 1;
+        finished = 0;
+        results_file = $results_path;
+        ansible_job_id = $local_jid;
+        _ansible_suppress_tmpdir_delete = $true;
+        ansible_async_watchdog_pid = $watchdog_pid
+    }
+
+    Write-AnsibleLog "INFO - writing initial async results to '$results_path'" "async_wrapper"
+    $result_json = ConvertTo-Json -InputObject $result -Depth 99 -Compress
+    Set-Content $results_path -Value $result_json
+
+    $np_timeout = $Payload.async_startup_timeout * 1000
+    Write-AnsibleLog "INFO - waiting for async process to connect to named pipe for $np_timeout milliseconds" "async_wrapper"
+    $wait_async = $pipe.BeginWaitForConnection($null, $null)
+    $wait_async.AsyncWaitHandle.WaitOne($np_timeout) > $null
+    if (-not $wait_async.IsCompleted) {
+        $msg = "Ansible encountered a timeout while waiting for the async task to start and connect to the named"
+        $msg += "pipe. This can be affected by the performance of the target - you can increase this timeout using"
+        $msg += "WIN_ASYNC_STARTUP_TIMEOUT or just for this host using the win_async_startup_timeout hostvar if "
+        $msg += "this keeps happening."
+        throw $msg
+    }
+    $pipe.EndWaitForConnection($wait_async)
+
+    Write-AnsibleLog "INFO - writing exec_wrapper and payload to async process" "async_wrapper"
+    $pipe.Write($payload_bytes, 0, $payload_bytes.Count)
+    $pipe.Flush()
+    $pipe.WaitForPipeDrain()
+}
+finally {
+    $pipe.Close()
+}
+
+Write-AnsibleLog "INFO - outputting initial async result: $result_json" "async_wrapper"
+Write-Output -InputObject $result_json
+Write-AnsibleLog "INFO - ending async_wrapper" "async_wrapper"
diff --git a/lib/ansible/executor/powershell/become_wrapper.ps1 b/lib/ansible/executor/powershell/become_wrapper.ps1
new file mode 100644
index 0000000..f40e265
--- /dev/null
+++ b/lib/ansible/executor/powershell/become_wrapper.ps1
@@ -0,0 +1,163 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+#Requires -Module Ansible.ModuleUtils.AddType
+#AnsibleRequires -CSharpUtil Ansible.AccessToken
+#AnsibleRequires -CSharpUtil Ansible.Become
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting become_wrapper" "become_wrapper"
+
+Function Get-EnumValue($enum, $flag_type, $value) {
+    $raw_enum_value = $value.Replace('_', '')
+    try {
+        $enum_value = [Enum]::Parse($enum, $raw_enum_value, $true)
+    }
+    catch [System.ArgumentException] {
+        $valid_options = [Enum]::GetNames($enum) | ForEach-Object -Process {
+            (($_ -creplace "(.)([A-Z][a-z]+)", '$1_$2') -creplace "([a-z0-9])([A-Z])", '$1_$2').ToString().ToLower()
+        }
+        throw "become_flags $flag_type value '$value' is not valid, valid values are: $($valid_options -join ", ")"
+    }
+    return $enum_value
+}
+
+Function Get-BecomeFlag($flags) {
+    $logon_type = [Ansible.AccessToken.LogonType]::Interactive
+    $logon_flags = [Ansible.Become.LogonFlags]::WithProfile
+
+    if ($null -eq $flags -or $flags -eq "") {
+        $flag_split = @()
+    }
+    elseif ($flags -is [string]) {
+        $flag_split = $flags.Split(" ")
+    }
+    else {
+        throw "become_flags must be a string, was $($flags.GetType())"
+    }
+
+    foreach ($flag in $flag_split) {
+        $split = $flag.Split("=")
+        if ($split.Count -ne 2) {
+            throw "become_flags entry '$flag' is in an invalid format, must be a key=value pair"
+        }
+        $flag_key = $split[0]
+        $flag_value = $split[1]
+        if ($flag_key -eq "logon_type") {
+            $enum_details = @{
+                enum = [Ansible.AccessToken.LogonType]
+                flag_type = $flag_key
+                value = $flag_value
+            }
+            $logon_type = Get-EnumValue @enum_details
+        }
+        elseif ($flag_key -eq "logon_flags") {
+            $logon_flag_values = $flag_value.Split(",")
+            $logon_flags = 0 -as [Ansible.Become.LogonFlags]
+            foreach ($logon_flag_value in $logon_flag_values) {
+                if ($logon_flag_value -eq "") {
+                    continue
+                }
+                $enum_details = @{
+                    enum = [Ansible.Become.LogonFlags]
+                    flag_type = $flag_key
+                    value = $logon_flag_value
+                }
+                $logon_flag = Get-EnumValue @enum_details
+                $logon_flags = $logon_flags -bor $logon_flag
+            }
+        }
+        else {
+            throw "become_flags key '$flag_key' is not a valid runas flag, must be 'logon_type' or 'logon_flags'"
+        }
+    }
+
+    return $logon_type, [Ansible.Become.LogonFlags]$logon_flags
+}
+
+Write-AnsibleLog "INFO - loading C# become code" "become_wrapper"
+$add_type_b64 = $Payload.powershell_modules["Ansible.ModuleUtils.AddType"]
+$add_type = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($add_type_b64))
+New-Module -Name Ansible.ModuleUtils.AddType -ScriptBlock ([ScriptBlock]::Create($add_type)) | Import-Module > $null
+
+$new_tmp = [System.Environment]::ExpandEnvironmentVariables($Payload.module_args["_ansible_remote_tmp"])
+$access_def = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.csharp_utils["Ansible.AccessToken"]))
+$become_def = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.csharp_utils["Ansible.Become"]))
+$process_def = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.csharp_utils["Ansible.Process"]))
+Add-CSharpType -References $access_def, $become_def, $process_def -TempPath $new_tmp -IncludeDebugInfo
+
+$username = $Payload.become_user
+$password = $Payload.become_password
+# We need to set password to the value of NullString so a null password is preserved when crossing the .NET
+# boundary. If we pass $null it will automatically be converted to "" and we need to keep the distinction for
+# accounts that don't have a password and when someone wants to become without knowing the password.
+if ($null -eq $password) {
+    $password = [NullString]::Value
+}
+
+try {
+    $logon_type, $logon_flags = Get-BecomeFlag -flags $Payload.become_flags
+}
+catch {
+    Write-AnsibleError -Message "internal error: failed to parse become_flags '$($Payload.become_flags)'" -ErrorRecord $_
+    $host.SetShouldExit(1)
+    return
+}
+Write-AnsibleLog "INFO - parsed become input, user: '$username', type: '$logon_type', flags: '$logon_flags'" "become_wrapper"
+
+# NB: CreateProcessWithTokenW commandline maxes out at 1024 chars, must
+# bootstrap via small wrapper which contains the exec_wrapper passed through the
+# stdin pipe. Cannot use 'powershell -' as the $ErrorActionPreference is always
+# set to Stop and cannot be changed. Also need to split the payload from the wrapper to prevent potentially
+# sensitive content from being logged by the scriptblock logger.
+$bootstrap_wrapper = {
+    &chcp.com 65001 > $null
+    $exec_wrapper_str = [System.Console]::In.ReadToEnd()
+    $split_parts = $exec_wrapper_str.Split(@("`0`0`0`0"), 2, [StringSplitOptions]::RemoveEmptyEntries)
+    Set-Variable -Name json_raw -Value $split_parts[1]
+    $exec_wrapper = [ScriptBlock]::Create($split_parts[0])
+    &$exec_wrapper
+}
+$exec_command = [System.Convert]::ToBase64String([System.Text.Encoding]::Unicode.GetBytes($bootstrap_wrapper.ToString()))
+$lp_command_line = "powershell.exe -NonInteractive -NoProfile -ExecutionPolicy Bypass -EncodedCommand $exec_command"
+$lp_current_directory = $env:SystemRoot  # TODO: should this be set to the become user's profile dir?
+
+# pop the become_wrapper action so we don't get stuck in a loop
+$Payload.actions = $Payload.actions[1..99]
+# we want the output from the exec_wrapper to be base64 encoded to preserve unicode chars
+$Payload.encoded_output = $true
+
+$payload_json = ConvertTo-Json -InputObject $Payload -Depth 99 -Compress
+# delimit the payload JSON from the wrapper to keep sensitive contents out of scriptblocks (which can be logged)
+$exec_wrapper = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.exec_wrapper))
+$exec_wrapper += "`0`0`0`0" + $payload_json
+
+try {
+    Write-AnsibleLog "INFO - starting become process '$lp_command_line'" "become_wrapper"
+    $result = [Ansible.Become.BecomeUtil]::CreateProcessAsUser($username, $password, $logon_flags, $logon_type,
+        $null, $lp_command_line, $lp_current_directory, $null, $exec_wrapper)
+    Write-AnsibleLog "INFO - become process complete with rc: $($result.ExitCode)" "become_wrapper"
+    $stdout = $result.StandardOut
+    try {
+        $stdout = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($stdout))
+    }
+    catch [FormatException] {
+        # output wasn't Base64, ignore as it may contain an error message we want to pass to Ansible
+        Write-AnsibleLog "WARN - become process stdout was not base64 encoded as expected: $stdout"
+    }
+
+    $host.UI.WriteLine($stdout)
+    $host.UI.WriteErrorLine($result.StandardError.Trim())
+    $host.SetShouldExit($result.ExitCode)
+}
+catch {
+    Write-AnsibleError -Message "internal error: failed to become user '$username'" -ErrorRecord $_
+    $host.SetShouldExit(1)
+}
+
+Write-AnsibleLog "INFO - ending become_wrapper" "become_wrapper"
diff --git a/lib/ansible/executor/powershell/bootstrap_wrapper.ps1 b/lib/ansible/executor/powershell/bootstrap_wrapper.ps1
new file mode 100644
index 0000000..cdba80c
--- /dev/null
+++ b/lib/ansible/executor/powershell/bootstrap_wrapper.ps1
@@ -0,0 +1,13 @@
+&chcp.com 65001 > $null
+
+if ($PSVersionTable.PSVersion -lt [Version]"3.0") {
+    '{"failed":true,"msg":"Ansible requires PowerShell v3.0 or newer"}'
+    exit 1
+}
+
+$exec_wrapper_str = $input | Out-String
+$split_parts = $exec_wrapper_str.Split(@("`0`0`0`0"), 2, [StringSplitOptions]::RemoveEmptyEntries)
+If (-not $split_parts.Length -eq 2) { throw "invalid payload" }
+Set-Variable -Name json_raw -Value $split_parts[1]
+$exec_wrapper = [ScriptBlock]::Create($split_parts[0])
+&$exec_wrapper
diff --git a/lib/ansible/executor/powershell/coverage_wrapper.ps1 b/lib/ansible/executor/powershell/coverage_wrapper.ps1
new file mode 100644
index 0000000..26cbe66
--- /dev/null
+++ b/lib/ansible/executor/powershell/coverage_wrapper.ps1
@@ -0,0 +1,199 @@
+# (c) 2019 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+#AnsibleRequires -Wrapper module_wrapper
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting coverage_wrapper" "coverage_wrapper"
+
+# Required to be set for psrp to we can set a breakpoint in the remote runspace
+if ($PSVersionTable.PSVersion -ge [Version]'4.0') {
+    $host.Runspace.Debugger.SetDebugMode([System.Management.Automation.DebugModes]::RemoteScript)
+}
+
+Function New-CoverageBreakpoint {
+    Param (
+        [String]$Path,
+        [ScriptBlock]$Code,
+        [String]$AnsiblePath
+    )
+
+    # It is quicker to pass in the code as a string instead of calling ParseFile as we already know the contents
+    $predicate = {
+        $args[0] -is [System.Management.Automation.Language.CommandBaseAst]
+    }
+    $script_cmds = $Code.Ast.FindAll($predicate, $true)
+
+    # Create an object that tracks the Ansible path of the file and the breakpoints that have been set in it
+    $info = [PSCustomObject]@{
+        Path = $AnsiblePath
+        Breakpoints = [System.Collections.Generic.List`1[System.Management.Automation.Breakpoint]]@()
+    }
+
+    # Keep track of lines that are already scanned. PowerShell can contains multiple commands in 1 line
+    $scanned_lines = [System.Collections.Generic.HashSet`1[System.Int32]]@()
+    foreach ($cmd in $script_cmds) {
+        if (-not $scanned_lines.Add($cmd.Extent.StartLineNumber)) {
+            continue
+        }
+
+        # Do not add any -Action value, even if it is $null or {}. Doing so will balloon the runtime.
+        $params = @{
+            Script = $Path
+            Line = $cmd.Extent.StartLineNumber
+            Column = $cmd.Extent.StartColumnNumber
+        }
+        $info.Breakpoints.Add((Set-PSBreakpoint @params))
+    }
+
+    $info
+}
+
+Function Compare-PathFilterPattern {
+    Param (
+        [String[]]$Patterns,
+        [String]$Path
+    )
+
+    foreach ($pattern in $Patterns) {
+        if ($Path -like $pattern) {
+            return $true
+        }
+    }
+    return $false
+}
+
+$module_name = $Payload.module_args["_ansible_module_name"]
+Write-AnsibleLog "INFO - building coverage payload for '$module_name'" "coverage_wrapper"
+
+# A PS Breakpoint needs an actual path to work properly, we create a temp directory that will store the module and
+# module_util code during execution
+$temp_path = Join-Path -Path ([System.IO.Path]::GetTempPath()) -ChildPath "ansible-coverage-$([System.IO.Path]::GetRandomFileName())"
+Write-AnsibleLog "INFO - Creating temp path for coverage files '$temp_path'" "coverage_wrapper"
+New-Item -Path $temp_path -ItemType Directory > $null
+$breakpoint_info = [System.Collections.Generic.List`1[PSObject]]@()
+
+# Ensures we create files with UTF-8 encoding and a BOM. This is critical to force the powershell engine to read files
+# as UTF-8 and not as the system's codepage.
+$file_encoding = 'UTF8'
+
+try {
+    $scripts = [System.Collections.Generic.List`1[System.Object]]@($script:common_functions)
+
+    $coverage_path_filter = $Payload.coverage.path_filter.Split(":", [StringSplitOptions]::RemoveEmptyEntries)
+
+    # We need to track what utils have already been added to the script for loading. This is because the load
+    # order is important and can have module_utils that rely on other utils.
+    $loaded_utils = [System.Collections.Generic.HashSet`1[System.String]]@()
+    $parse_util = {
+        $util_name = $args[0]
+        if (-not $loaded_utils.Add($util_name)) {
+            return
+        }
+
+        $util_code = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.powershell_modules.$util_name))
+        $util_sb = [ScriptBlock]::Create($util_code)
+        $util_path = Join-Path -Path $temp_path -ChildPath "$($util_name).psm1"
+
+        Write-AnsibleLog "INFO - Outputting module_util $util_name to temp file '$util_path'" "coverage_wrapper"
+        Set-Content -LiteralPath $util_path -Value $util_code -Encoding $file_encoding
+
+        $ansible_path = $Payload.coverage.module_util_paths.$util_name
+        if ((Compare-PathFilterPattern -Patterns $coverage_path_filter -Path $ansible_path)) {
+            $cov_params = @{
+                Path = $util_path
+                Code = $util_sb
+                AnsiblePath = $ansible_path
+            }
+            $breakpoints = New-CoverageBreakpoint @cov_params
+            $breakpoint_info.Add($breakpoints)
+        }
+
+        if ($null -ne $util_sb.Ast.ScriptRequirements) {
+            foreach ($required_util in $util_sb.Ast.ScriptRequirements.RequiredModules) {
+                &$parse_util $required_util.Name
+            }
+        }
+        Write-AnsibleLog "INFO - Adding util $util_name to scripts to run" "coverage_wrapper"
+        $scripts.Add("Import-Module -Name '$util_path'")
+    }
+    foreach ($util in $Payload.powershell_modules.Keys) {
+        &$parse_util $util
+    }
+
+    $module = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.module_entry))
+    $module_path = Join-Path -Path $temp_path -ChildPath "$($module_name).ps1"
+    Write-AnsibleLog "INFO - Ouputting module $module_name to temp file '$module_path'" "coverage_wrapper"
+    Set-Content -LiteralPath $module_path -Value $module -Encoding $file_encoding
+    $scripts.Add($module_path)
+
+    $ansible_path = $Payload.coverage.module_path
+    if ((Compare-PathFilterPattern -Patterns $coverage_path_filter -Path $ansible_path)) {
+        $cov_params = @{
+            Path = $module_path
+            Code = [ScriptBlock]::Create($module)
+            AnsiblePath = $Payload.coverage.module_path
+        }
+        $breakpoints = New-CoverageBreakpoint @cov_params
+        $breakpoint_info.Add($breakpoints)
+    }
+
+    $variables = [System.Collections.ArrayList]@(@{ Name = "complex_args"; Value = $Payload.module_args; Scope = "Global" })
+    $entrypoint = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($payload.module_wrapper))
+    $entrypoint = [ScriptBlock]::Create($entrypoint)
+
+    $params = @{
+        Scripts = $scripts
+        Variables = $variables
+        Environment = $Payload.environment
+        ModuleName = $module_name
+    }
+    if ($breakpoint_info) {
+        $params.Breakpoints = $breakpoint_info.Breakpoints
+    }
+
+    try {
+        &$entrypoint @params
+    }
+    finally {
+        # Processing here is kept to an absolute minimum to make sure each task runtime is kept as small as
+        # possible. Once all the tests have been run ansible-test will collect this info and process it locally in
+        # one go.
+        Write-AnsibleLog "INFO - Creating coverage result output" "coverage_wrapper"
+        $coverage_info = @{}
+        foreach ($info in $breakpoint_info) {
+            $coverage_info.($info.Path) = $info.Breakpoints | Select-Object -Property Line, HitCount
+        }
+
+        # The coverage.output value is a filename set by the Ansible controller. We append some more remote side
+        # info to the filename to make it unique and identify the remote host a bit more.
+        $ps_version = "$($PSVersionTable.PSVersion.Major).$($PSVersionTable.PSVersion.Minor)"
+        $coverage_output_path = "$($Payload.coverage.output)=powershell-$ps_version=coverage.$($env:COMPUTERNAME).$PID.$(Get-Random)"
+        $code_cov_json = ConvertTo-Json -InputObject $coverage_info -Compress
+
+        Write-AnsibleLog "INFO - Outputting coverage json to '$coverage_output_path'" "coverage_wrapper"
+        # Ansible controller expects these files to be UTF-8 without a BOM, use .NET for this.
+        $utf8_no_bom = New-Object -TypeName System.Text.UTF8Encoding -ArgumentList $false
+        [System.IO.File]::WriteAllbytes($coverage_output_path, $utf8_no_bom.GetBytes($code_cov_json))
+    }
+}
+finally {
+    try {
+        if ($breakpoint_info) {
+            foreach ($b in $breakpoint_info.Breakpoints) {
+                Remove-PSBreakpoint -Breakpoint $b
+            }
+        }
+    }
+    finally {
+        Write-AnsibleLog "INFO - Remove temp coverage folder '$temp_path'" "coverage_wrapper"
+        Remove-Item -LiteralPath $temp_path -Force -Recurse
+    }
+}
+
+Write-AnsibleLog "INFO - ending coverage_wrapper" "coverage_wrapper"
diff --git a/lib/ansible/executor/powershell/exec_wrapper.ps1 b/lib/ansible/executor/powershell/exec_wrapper.ps1
new file mode 100644
index 0000000..0f97bdf
--- /dev/null
+++ b/lib/ansible/executor/powershell/exec_wrapper.ps1
@@ -0,0 +1,237 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+begin {
+    $DebugPreference = "Continue"
+    $ProgressPreference = "SilentlyContinue"
+    $ErrorActionPreference = "Stop"
+    Set-StrictMode -Version 2
+
+    # common functions that are loaded in exec and module context, this is set
+    # as a script scoped variable so async_watchdog and module_wrapper can
+    # access the functions when creating their Runspaces
+    $script:common_functions = {
+        Function ConvertFrom-AnsibleJson {
+            <#
+            .SYNOPSIS
+            Converts a JSON string to a Hashtable/Array in the fastest way
+            possible. Unfortunately ConvertFrom-Json is still faster but outputs
+            a PSCustomObject which is combersone for module consumption.
+
+            .PARAMETER InputObject
+            [String] The JSON string to deserialize.
+            #>
+            param(
+                [Parameter(Mandatory = $true, Position = 0)][String]$InputObject
+            )
+
+            # we can use -AsHashtable to get PowerShell to convert the JSON to
+            # a Hashtable and not a PSCustomObject. This was added in PowerShell
+            # 6.0, fall back to a manual conversion for older versions
+            $cmdlet = Get-Command -Name ConvertFrom-Json -CommandType Cmdlet
+            if ("AsHashtable" -in $cmdlet.Parameters.Keys) {
+                return , (ConvertFrom-Json -InputObject $InputObject -AsHashtable)
+            }
+            else {
+                # get the PSCustomObject and then manually convert from there
+                $raw_obj = ConvertFrom-Json -InputObject $InputObject
+
+                Function ConvertTo-Hashtable {
+                    param($InputObject)
+
+                    if ($null -eq $InputObject) {
+                        return $null
+                    }
+
+                    if ($InputObject -is [PSCustomObject]) {
+                        $new_value = @{}
+                        foreach ($prop in $InputObject.PSObject.Properties.GetEnumerator()) {
+                            $new_value.($prop.Name) = (ConvertTo-Hashtable -InputObject $prop.Value)
+                        }
+                        return , $new_value
+                    }
+                    elseif ($InputObject -is [Array]) {
+                        $new_value = [System.Collections.ArrayList]@()
+                        foreach ($val in $InputObject) {
+                            $new_value.Add((ConvertTo-Hashtable -InputObject $val)) > $null
+                        }
+                        return , $new_value.ToArray()
+                    }
+                    else {
+                        return , $InputObject
+                    }
+                }
+                return , (ConvertTo-Hashtable -InputObject $raw_obj)
+            }
+        }
+
+        Function Format-AnsibleException {
+            <#
+            .SYNOPSIS
+            Formats a PowerShell ErrorRecord to a string that's fit for human
+            consumption.
+
+            .NOTES
+            Using Out-String can give us the first part of the exception but it
+            also wraps the messages at 80 chars which is not ideal. We also
+            append the ScriptStackTrace and the .NET StackTrace if present.
+            #>
+            param([System.Management.Automation.ErrorRecord]$ErrorRecord)
+
+            $exception = @"
+$($ErrorRecord.ToString())
+$($ErrorRecord.InvocationInfo.PositionMessage)
+    + CategoryInfo          : $($ErrorRecord.CategoryInfo.ToString())
+    + FullyQualifiedErrorId : $($ErrorRecord.FullyQualifiedErrorId.ToString())
+"@
+            # module_common strip comments and empty newlines, need to manually
+            # add a preceding newline using `r`n
+            $exception += "`r`n`r`nScriptStackTrace:`r`n$($ErrorRecord.ScriptStackTrace)`r`n"
+
+            # exceptions from C# will also have a StackTrace which we
+            # append if found
+            if ($null -ne $ErrorRecord.Exception.StackTrace) {
+                $exception += "`r`n$($ErrorRecord.Exception.ToString())"
+            }
+
+            return $exception
+        }
+    }
+    .$common_functions
+
+    # common wrapper functions used in the exec wrappers, this is defined in a
+    # script scoped variable so async_watchdog can pass them into the async job
+    $script:wrapper_functions = {
+        Function Write-AnsibleError {
+            <#
+            .SYNOPSIS
+            Writes an error message to a JSON string in the format that Ansible
+            understands. Also optionally adds an exception record if the
+            ErrorRecord is passed through.
+            #>
+            param(
+                [Parameter(Mandatory = $true)][String]$Message,
+                [System.Management.Automation.ErrorRecord]$ErrorRecord = $null
+            )
+            $result = @{
+                msg = $Message
+                failed = $true
+            }
+            if ($null -ne $ErrorRecord) {
+                $result.msg += ": $($ErrorRecord.Exception.Message)"
+                $result.exception = (Format-AnsibleException -ErrorRecord $ErrorRecord)
+            }
+            Write-Output -InputObject (ConvertTo-Json -InputObject $result -Depth 99 -Compress)
+        }
+
+        Function Write-AnsibleLog {
+            <#
+            .SYNOPSIS
+            Used as a debugging tool to log events to a file as they run in the
+            exec wrappers. By default this is a noop function but the $log_path
+            can be manually set to enable it. Manually set ANSIBLE_EXEC_DEBUG as
+            an env value on the Windows host that this is run on to enable.
+            #>
+            param(
+                [Parameter(Mandatory = $true, Position = 0)][String]$Message,
+                [Parameter(Position = 1)][String]$Wrapper
+            )
+
+            $log_path = $env:ANSIBLE_EXEC_DEBUG
+            if ($log_path) {
+                $log_path = [System.Environment]::ExpandEnvironmentVariables($log_path)
+                $parent_path = [System.IO.Path]::GetDirectoryName($log_path)
+                if (Test-Path -LiteralPath $parent_path -PathType Container) {
+                    $msg = "{0:u} - {1} - {2} - " -f (Get-Date), $pid, ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name)
+                    if ($null -ne $Wrapper) {
+                        $msg += "$Wrapper - "
+                    }
+                    $msg += $Message + "`r`n"
+                    $msg_bytes = [System.Text.Encoding]::UTF8.GetBytes($msg)
+
+                    $fs = [System.IO.File]::Open($log_path, [System.IO.FileMode]::Append,
+                        [System.IO.FileAccess]::Write, [System.IO.FileShare]::ReadWrite)
+                    try {
+                        $fs.Write($msg_bytes, 0, $msg_bytes.Length)
+                    }
+                    finally {
+                        $fs.Close()
+                    }
+                }
+            }
+        }
+    }
+    .$wrapper_functions
+
+    # only init and stream in $json_raw if it wasn't set by the enclosing scope
+    if (-not $(Get-Variable "json_raw" -ErrorAction SilentlyContinue)) {
+        $json_raw = ''
+    }
+} process {
+    $json_raw += [String]$input
+} end {
+    Write-AnsibleLog "INFO - starting exec_wrapper" "exec_wrapper"
+    if (-not $json_raw) {
+        Write-AnsibleError -Message "internal error: no input given to PowerShell exec wrapper"
+        exit 1
+    }
+
+    Write-AnsibleLog "INFO - converting json raw to a payload" "exec_wrapper"
+    $payload = ConvertFrom-AnsibleJson -InputObject $json_raw
+
+    # TODO: handle binary modules
+    # TODO: handle persistence
+
+    if ($payload.min_os_version) {
+        $min_os_version = [Version]$payload.min_os_version
+        # Environment.OSVersion.Version is deprecated and may not return the
+        # right version
+        $actual_os_version = [Version](Get-Item -Path $env:SystemRoot\System32\kernel32.dll).VersionInfo.ProductVersion
+
+        Write-AnsibleLog "INFO - checking if actual os version '$actual_os_version' is less than the min os version '$min_os_version'" "exec_wrapper"
+        if ($actual_os_version -lt $min_os_version) {
+            $msg = "internal error: This module cannot run on this OS as it requires a minimum version of $min_os_version, actual was $actual_os_version"
+            Write-AnsibleError -Message $msg
+            exit 1
+        }
+    }
+    if ($payload.min_ps_version) {
+        $min_ps_version = [Version]$payload.min_ps_version
+        $actual_ps_version = $PSVersionTable.PSVersion
+
+        Write-AnsibleLog "INFO - checking if actual PS version '$actual_ps_version' is less than the min PS version '$min_ps_version'" "exec_wrapper"
+        if ($actual_ps_version -lt $min_ps_version) {
+            $msg = "internal error: This module cannot run as it requires a minimum PowerShell version of $min_ps_version, actual was $actual_ps_version"
+            Write-AnsibleError -Message $msg
+            exit 1
+        }
+    }
+
+    # pop 0th action as entrypoint
+    $action = $payload.actions[0]
+    Write-AnsibleLog "INFO - running action $action" "exec_wrapper"
+
+    $entrypoint = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($payload.($action)))
+    $entrypoint = [ScriptBlock]::Create($entrypoint)
+    # so we preserve the formatting and don't fall prey to locale issues, some
+    # wrappers want the output to be in base64 form, we store the value here in
+    # case the wrapper changes the value when they create a payload for their
+    # own exec_wrapper
+    $encoded_output = $payload.encoded_output
+
+    try {
+        $output = &$entrypoint -Payload $payload
+        if ($encoded_output -and $null -ne $output) {
+            $b64_output = [System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes($output))
+            Write-Output -InputObject $b64_output
+        }
+        else {
+            $output
+        }
+    }
+    catch {
+        Write-AnsibleError -Message "internal error: failed to run exec_wrapper action $action" -ErrorRecord $_
+        exit 1
+    }
+    Write-AnsibleLog "INFO - ending exec_wrapper" "exec_wrapper"
+}
diff --git a/lib/ansible/executor/powershell/module_manifest.py b/lib/ansible/executor/powershell/module_manifest.py
new file mode 100644
index 0000000..970e848
--- /dev/null
+++ b/lib/ansible/executor/powershell/module_manifest.py
@@ -0,0 +1,402 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import base64
+import errno
+import json
+import os
+import pkgutil
+import random
+import re
+
+from ansible.module_utils.compat.version import LooseVersion
+
+from ansible import constants as C
+from ansible.errors import AnsibleError
+from ansible.module_utils._text import to_bytes, to_native, to_text
+from ansible.module_utils.compat.importlib import import_module
+from ansible.plugins.loader import ps_module_utils_loader
+from ansible.utils.collection_loader import resource_from_fqcr
+
+
+class PSModuleDepFinder(object):
+
+    def __init__(self):
+        # This is also used by validate-modules to get a module's required utils in base and a collection.
+        self.ps_modules = dict()
+        self.exec_scripts = dict()
+
+        # by defining an explicit dict of cs utils and where they are used, we
+        # can potentially save time by not adding the type multiple times if it
+        # isn't needed
+        self.cs_utils_wrapper = dict()
+        self.cs_utils_module = dict()
+
+        self.ps_version = None
+        self.os_version = None
+        self.become = False
+
+        self._re_cs_module = [
+            # Reference C# module_util in another C# util, this must always be the fully qualified name.
+            # 'using ansible_collections.{namespace}.{collection}.plugins.module_utils.{name}'
+            re.compile(to_bytes(r'(?i)^using\s((Ansible\..+)|'
+                                r'(ansible_collections\.\w+\.\w+\.plugins\.module_utils\.[\w\.]+));\s*$')),
+        ]
+
+        self._re_cs_in_ps_module = [
+            # Reference C# module_util in a PowerShell module
+            # '#AnsibleRequires -CSharpUtil Ansible.{name}'
+            # '#AnsibleRequires -CSharpUtil ansible_collections.{namespace}.{collection}.plugins.module_utils.{name}'
+            # '#AnsibleRequires -CSharpUtil ..module_utils.{name}'
+            # Can have '-Optional' at the end to denote the util is optional
+            re.compile(to_bytes(r'(?i)^#\s*ansiblerequires\s+-csharputil\s+((Ansible\.[\w\.]+)|'
+                                r'(ansible_collections\.\w+\.\w+\.plugins\.module_utils\.[\w\.]+)|'
+                                r'(\.[\w\.]+))(?P<optional>\s+-Optional){0,1}')),
+        ]
+
+        self._re_ps_module = [
+            # Original way of referencing a builtin module_util
+            # '#Requires -Module Ansible.ModuleUtils.{name}
+            re.compile(to_bytes(r'(?i)^#\s*requires\s+\-module(?:s?)\s*(Ansible\.ModuleUtils\..+)')),
+            # New way of referencing a builtin and collection module_util
+            # '#AnsibleRequires -PowerShell Ansible.ModuleUtils.{name}'
+            # '#AnsibleRequires -PowerShell ansible_collections.{namespace}.{collection}.plugins.module_utils.{name}'
+            # '#AnsibleRequires -PowerShell ..module_utils.{name}'
+            # Can have '-Optional' at the end to denote the util is optional
+            re.compile(to_bytes(r'(?i)^#\s*ansiblerequires\s+-powershell\s+((Ansible\.ModuleUtils\.[\w\.]+)|'
+                                r'(ansible_collections\.\w+\.\w+\.plugins\.module_utils\.[\w\.]+)|'
+                                r'(\.[\w\.]+))(?P<optional>\s+-Optional){0,1}')),
+        ]
+
+        self._re_wrapper = re.compile(to_bytes(r'(?i)^#\s*ansiblerequires\s+-wrapper\s+(\w*)'))
+        self._re_ps_version = re.compile(to_bytes(r'(?i)^#requires\s+\-version\s+([0-9]+(\.[0-9]+){0,3})$'))
+        self._re_os_version = re.compile(to_bytes(r'(?i)^#ansiblerequires\s+\-osversion\s+([0-9]+(\.[0-9]+){0,3})$'))
+        self._re_become = re.compile(to_bytes(r'(?i)^#ansiblerequires\s+\-become$'))
+
+    def scan_module(self, module_data, fqn=None, wrapper=False, powershell=True):
+        lines = module_data.split(b'\n')
+        module_utils = set()
+        if wrapper:
+            cs_utils = self.cs_utils_wrapper
+        else:
+            cs_utils = self.cs_utils_module
+
+        if powershell:
+            checks = [
+                # PS module contains '#Requires -Module Ansible.ModuleUtils.*'
+                # PS module contains '#AnsibleRequires -Powershell Ansible.*' (or collections module_utils ref)
+                (self._re_ps_module, self.ps_modules, ".psm1"),
+                # PS module contains '#AnsibleRequires -CSharpUtil Ansible.*' (or collections module_utils ref)
+                (self._re_cs_in_ps_module, cs_utils, ".cs"),
+            ]
+        else:
+            checks = [
+                # CS module contains 'using Ansible.*;' or 'using ansible_collections.ns.coll.plugins.module_utils.*;'
+                (self._re_cs_module, cs_utils, ".cs"),
+            ]
+
+        for line in lines:
+            for check in checks:
+                for pattern in check[0]:
+                    match = pattern.match(line)
+                    if match:
+                        # tolerate windows line endings by stripping any remaining
+                        # newline chars
+                        module_util_name = to_text(match.group(1).rstrip())
+                        match_dict = match.groupdict()
+                        optional = match_dict.get('optional', None) is not None
+
+                        if module_util_name not in check[1].keys():
+                            module_utils.add((module_util_name, check[2], fqn, optional))
+
+                        break
+
+            if powershell:
+                ps_version_match = self._re_ps_version.match(line)
+                if ps_version_match:
+                    self._parse_version_match(ps_version_match, "ps_version")
+
+                os_version_match = self._re_os_version.match(line)
+                if os_version_match:
+                    self._parse_version_match(os_version_match, "os_version")
+
+                # once become is set, no need to keep on checking recursively
+                if not self.become:
+                    become_match = self._re_become.match(line)
+                    if become_match:
+                        self.become = True
+
+            if wrapper:
+                wrapper_match = self._re_wrapper.match(line)
+                if wrapper_match:
+                    self.scan_exec_script(wrapper_match.group(1).rstrip())
+
+        # recursively drill into each Requires to see if there are any more
+        # requirements
+        for m in set(module_utils):
+            self._add_module(*m, wrapper=wrapper)
+
+    def scan_exec_script(self, name):
+        # scans lib/ansible/executor/powershell for scripts used in the module
+        # exec side. It also scans these scripts for any dependencies
+        name = to_text(name)
+        if name in self.exec_scripts.keys():
+            return
+
+        data = pkgutil.get_data("ansible.executor.powershell", to_native(name + ".ps1"))
+        if data is None:
+            raise AnsibleError("Could not find executor powershell script "
+                               "for '%s'" % name)
+
+        b_data = to_bytes(data)
+
+        # remove comments to reduce the payload size in the exec wrappers
+        if C.DEFAULT_DEBUG:
+            exec_script = b_data
+        else:
+            exec_script = _strip_comments(b_data)
+        self.exec_scripts[name] = to_bytes(exec_script)
+        self.scan_module(b_data, wrapper=True, powershell=True)
+
+    def _add_module(self, name, ext, fqn, optional, wrapper=False):
+        m = to_text(name)
+
+        util_fqn = None
+
+        if m.startswith("Ansible."):
+            # Builtin util, use plugin loader to get the data
+            mu_path = ps_module_utils_loader.find_plugin(m, ext)
+
+            if not mu_path:
+                if optional:
+                    return
+
+                raise AnsibleError('Could not find imported module support code '
+                                   'for \'%s\'' % m)
+
+            module_util_data = to_bytes(_slurp(mu_path))
+        else:
+            # Collection util, load the package data based on the util import.
+
+            submodules = m.split(".")
+            if m.startswith('.'):
+                fqn_submodules = fqn.split('.')
+                for submodule in submodules:
+                    if submodule:
+                        break
+                    del fqn_submodules[-1]
+
+                submodules = fqn_submodules + [s for s in submodules if s]
+
+            n_package_name = to_native('.'.join(submodules[:-1]), errors='surrogate_or_strict')
+            n_resource_name = to_native(submodules[-1] + ext, errors='surrogate_or_strict')
+
+            try:
+                module_util = import_module(n_package_name)
+                pkg_data = pkgutil.get_data(n_package_name, n_resource_name)
+                if pkg_data is None:
+                    raise ImportError("No package data found")
+
+                module_util_data = to_bytes(pkg_data, errors='surrogate_or_strict')
+                util_fqn = to_text("%s.%s " % (n_package_name, submodules[-1]), errors='surrogate_or_strict')
+
+                # Get the path of the util which is required for coverage collection.
+                resource_paths = list(module_util.__path__)
+                if len(resource_paths) != 1:
+                    # This should never happen with a collection but we are just being defensive about it.
+                    raise AnsibleError("Internal error: Referenced module_util package '%s' contains 0 or multiple "
+                                       "import locations when we only expect 1." % n_package_name)
+                mu_path = os.path.join(resource_paths[0], n_resource_name)
+            except (ImportError, OSError) as err:
+                if getattr(err, "errno", errno.ENOENT) == errno.ENOENT:
+                    if optional:
+                        return
+
+                    raise AnsibleError('Could not find collection imported module support code for \'%s\''
+                                       % to_native(m))
+
+                else:
+                    raise
+
+        util_info = {
+            'data': module_util_data,
+            'path': to_text(mu_path),
+        }
+        if ext == ".psm1":
+            self.ps_modules[m] = util_info
+        else:
+            if wrapper:
+                self.cs_utils_wrapper[m] = util_info
+            else:
+                self.cs_utils_module[m] = util_info
+        self.scan_module(module_util_data, fqn=util_fqn, wrapper=wrapper, powershell=(ext == ".psm1"))
+
+    def _parse_version_match(self, match, attribute):
+        new_version = to_text(match.group(1)).rstrip()
+
+        # PowerShell cannot cast a string of "1" to Version, it must have at
+        # least the major.minor for it to be valid so we append 0
+        if match.group(2) is None:
+            new_version = "%s.0" % new_version
+
+        existing_version = getattr(self, attribute, None)
+        if existing_version is None:
+            setattr(self, attribute, new_version)
+        else:
+            # determine which is the latest version and set that
+            if LooseVersion(new_version) > LooseVersion(existing_version):
+                setattr(self, attribute, new_version)
+
+
+def _slurp(path):
+    if not os.path.exists(path):
+        raise AnsibleError("imported module support code does not exist at %s"
+                           % os.path.abspath(path))
+    fd = open(path, 'rb')
+    data = fd.read()
+    fd.close()
+    return data
+
+
+def _strip_comments(source):
+    # Strip comments and blank lines from the wrapper
+    buf = []
+    start_block = False
+    for line in source.splitlines():
+        l = line.strip()
+
+        if start_block and l.endswith(b'#>'):
+            start_block = False
+            continue
+        elif start_block:
+            continue
+        elif l.startswith(b'<#'):
+            start_block = True
+            continue
+        elif not l or l.startswith(b'#'):
+            continue
+
+        buf.append(line)
+    return b'\n'.join(buf)
+
+
+def _create_powershell_wrapper(b_module_data, module_path, module_args,
+                               environment, async_timeout, become,
+                               become_method, become_user, become_password,
+                               become_flags, substyle, task_vars, module_fqn):
+    # creates the manifest/wrapper used in PowerShell/C# modules to enable
+    # things like become and async - this is also called in action/script.py
+
+    # FUTURE: add process_wrapper.ps1 to run module_wrapper in a new process
+    # if running under a persistent connection and substyle is C# so we
+    # don't have type conflicts
+    finder = PSModuleDepFinder()
+    if substyle != 'script':
+        # don't scan the module for util dependencies and other Ansible related
+        # flags if the substyle is 'script' which is set by action/script
+        finder.scan_module(b_module_data, fqn=module_fqn, powershell=(substyle == "powershell"))
+
+    module_wrapper = "module_%s_wrapper" % substyle
+    exec_manifest = dict(
+        module_entry=to_text(base64.b64encode(b_module_data)),
+        powershell_modules=dict(),
+        csharp_utils=dict(),
+        csharp_utils_module=list(),  # csharp_utils only required by a module
+        module_args=module_args,
+        actions=[module_wrapper],
+        environment=environment,
+        encoded_output=False,
+    )
+    finder.scan_exec_script(module_wrapper)
+
+    if async_timeout > 0:
+        finder.scan_exec_script('exec_wrapper')
+        finder.scan_exec_script('async_watchdog')
+        finder.scan_exec_script('async_wrapper')
+
+        exec_manifest["actions"].insert(0, 'async_watchdog')
+        exec_manifest["actions"].insert(0, 'async_wrapper')
+        exec_manifest["async_jid"] = str(random.randint(0, 999999999999))
+        exec_manifest["async_timeout_sec"] = async_timeout
+        exec_manifest["async_startup_timeout"] = C.config.get_config_value("WIN_ASYNC_STARTUP_TIMEOUT", variables=task_vars)
+
+    if become and resource_from_fqcr(become_method) == 'runas':  # runas and namespace.collection.runas
+        finder.scan_exec_script('exec_wrapper')
+        finder.scan_exec_script('become_wrapper')
+
+        exec_manifest["actions"].insert(0, 'become_wrapper')
+        exec_manifest["become_user"] = become_user
+        exec_manifest["become_password"] = become_password
+        exec_manifest['become_flags'] = become_flags
+
+    exec_manifest['min_ps_version'] = finder.ps_version
+    exec_manifest['min_os_version'] = finder.os_version
+    if finder.become and 'become_wrapper' not in exec_manifest['actions']:
+        finder.scan_exec_script('exec_wrapper')
+        finder.scan_exec_script('become_wrapper')
+
+        exec_manifest['actions'].insert(0, 'become_wrapper')
+        exec_manifest['become_user'] = 'SYSTEM'
+        exec_manifest['become_password'] = None
+        exec_manifest['become_flags'] = None
+
+    coverage_manifest = dict(
+        module_path=module_path,
+        module_util_paths=dict(),
+        output=None,
+    )
+    coverage_output = C.config.get_config_value('COVERAGE_REMOTE_OUTPUT', variables=task_vars)
+    if coverage_output and substyle == 'powershell':
+        finder.scan_exec_script('coverage_wrapper')
+        coverage_manifest['output'] = coverage_output
+
+        coverage_enabled = C.config.get_config_value('COVERAGE_REMOTE_PATHS', variables=task_vars)
+        coverage_manifest['path_filter'] = coverage_enabled
+
+    # make sure Ansible.ModuleUtils.AddType is added if any C# utils are used
+    if len(finder.cs_utils_wrapper) > 0 or len(finder.cs_utils_module) > 0:
+        finder._add_module(b"Ansible.ModuleUtils.AddType", ".psm1", None, False,
+                           wrapper=False)
+
+    # exec_wrapper is only required to be part of the payload if using
+    # become or async, to save on payload space we check if exec_wrapper has
+    # already been added, and remove it manually if it hasn't later
+    exec_required = "exec_wrapper" in finder.exec_scripts.keys()
+    finder.scan_exec_script("exec_wrapper")
+    # must contain an empty newline so it runs the begin/process/end block
+    finder.exec_scripts["exec_wrapper"] += b"\n\n"
+
+    exec_wrapper = finder.exec_scripts["exec_wrapper"]
+    if not exec_required:
+        finder.exec_scripts.pop("exec_wrapper")
+
+    for name, data in finder.exec_scripts.items():
+        b64_data = to_text(base64.b64encode(data))
+        exec_manifest[name] = b64_data
+
+    for name, data in finder.ps_modules.items():
+        b64_data = to_text(base64.b64encode(data['data']))
+        exec_manifest['powershell_modules'][name] = b64_data
+        coverage_manifest['module_util_paths'][name] = data['path']
+
+    cs_utils = {}
+    for cs_util in [finder.cs_utils_wrapper, finder.cs_utils_module]:
+        for name, data in cs_util.items():
+            cs_utils[name] = data['data']
+
+    for name, data in cs_utils.items():
+        b64_data = to_text(base64.b64encode(data))
+        exec_manifest['csharp_utils'][name] = b64_data
+    exec_manifest['csharp_utils_module'] = list(finder.cs_utils_module.keys())
+
+    # To save on the data we are sending across we only add the coverage info if coverage is being run
+    if 'coverage_wrapper' in exec_manifest:
+        exec_manifest['coverage'] = coverage_manifest
+
+    b_json = to_bytes(json.dumps(exec_manifest))
+    # delimit the payload JSON from the wrapper to keep sensitive contents out of scriptblocks (which can be logged)
+    b_data = exec_wrapper + b'\0\0\0\0' + b_json
+    return b_data
diff --git a/lib/ansible/executor/powershell/module_powershell_wrapper.ps1 b/lib/ansible/executor/powershell/module_powershell_wrapper.ps1
new file mode 100644
index 0000000..c35c84c
--- /dev/null
+++ b/lib/ansible/executor/powershell/module_powershell_wrapper.ps1
@@ -0,0 +1,75 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+#AnsibleRequires -Wrapper module_wrapper
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting module_powershell_wrapper" "module_powershell_wrapper"
+
+$module_name = $Payload.module_args["_ansible_module_name"]
+Write-AnsibleLog "INFO - building module payload for '$module_name'" "module_powershell_wrapper"
+
+# compile any C# module utils passed in from the controller, Add-CSharpType is
+# automatically added to the payload manifest if any csharp util is set
+$csharp_utils = [System.Collections.ArrayList]@()
+foreach ($csharp_util in $Payload.csharp_utils_module) {
+    Write-AnsibleLog "INFO - adding $csharp_util to list of C# references to compile" "module_powershell_wrapper"
+    $util_code = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.csharp_utils[$csharp_util]))
+    $csharp_utils.Add($util_code) > $null
+}
+if ($csharp_utils.Count -gt 0) {
+    $add_type_b64 = $Payload.powershell_modules["Ansible.ModuleUtils.AddType"]
+    $add_type = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($add_type_b64))
+    New-Module -Name Ansible.ModuleUtils.AddType -ScriptBlock ([ScriptBlock]::Create($add_type)) | Import-Module > $null
+
+    # add any C# references so the module does not have to do so
+    $new_tmp = [System.Environment]::ExpandEnvironmentVariables($Payload.module_args["_ansible_remote_tmp"])
+    Add-CSharpType -References $csharp_utils -TempPath $new_tmp -IncludeDebugInfo
+}
+
+if ($Payload.ContainsKey("coverage") -and $null -ne $host.Runspace -and $null -ne $host.Runspace.Debugger) {
+    $entrypoint = $payload.coverage_wrapper
+
+    $params = @{
+        Payload = $Payload
+    }
+}
+else {
+    # get the common module_wrapper code and invoke that to run the module
+    $module = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.module_entry))
+    $variables = [System.Collections.ArrayList]@(@{ Name = "complex_args"; Value = $Payload.module_args; Scope = "Global" })
+    $entrypoint = $Payload.module_wrapper
+
+    $params = @{
+        Scripts = @($script:common_functions, $module)
+        Variables = $variables
+        Environment = $Payload.environment
+        Modules = $Payload.powershell_modules
+        ModuleName = $module_name
+    }
+}
+
+$entrypoint = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($entrypoint))
+$entrypoint = [ScriptBlock]::Create($entrypoint)
+
+try {
+    &$entrypoint @params
+}
+catch {
+    # failed to invoke the PowerShell module, capture the exception and
+    # output a pretty error for Ansible to parse
+    $result = @{
+        msg = "Failed to invoke PowerShell module: $($_.Exception.Message)"
+        failed = $true
+        exception = (Format-AnsibleException -ErrorRecord $_)
+    }
+    Write-Output -InputObject (ConvertTo-Json -InputObject $result -Depth 99 -Compress)
+    $host.SetShouldExit(1)
+}
+
+Write-AnsibleLog "INFO - ending module_powershell_wrapper" "module_powershell_wrapper"
diff --git a/lib/ansible/executor/powershell/module_script_wrapper.ps1 b/lib/ansible/executor/powershell/module_script_wrapper.ps1
new file mode 100644
index 0000000..dd8420f
--- /dev/null
+++ b/lib/ansible/executor/powershell/module_script_wrapper.ps1
@@ -0,0 +1,22 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+param(
+    [Parameter(Mandatory = $true)][System.Collections.IDictionary]$Payload
+)
+
+#AnsibleRequires -Wrapper module_wrapper
+
+$ErrorActionPreference = "Stop"
+
+Write-AnsibleLog "INFO - starting module_script_wrapper" "module_script_wrapper"
+
+$script = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($Payload.module_entry))
+
+# get the common module_wrapper code and invoke that to run the module
+$entrypoint = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($payload.module_wrapper))
+$entrypoint = [ScriptBlock]::Create($entrypoint)
+
+&$entrypoint -Scripts $script -Environment $Payload.environment -ModuleName "script"
+
+Write-AnsibleLog "INFO - ending module_script_wrapper" "module_script_wrapper"
diff --git a/lib/ansible/executor/powershell/module_wrapper.ps1 b/lib/ansible/executor/powershell/module_wrapper.ps1
new file mode 100644
index 0000000..20a9677
--- /dev/null
+++ b/lib/ansible/executor/powershell/module_wrapper.ps1
@@ -0,0 +1,226 @@
+# (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+<#
+.SYNOPSIS
+Invokes an Ansible module in a new Runspace. This cmdlet will output the
+module's output and write any errors to the error stream of the current
+host.
+
+.PARAMETER Scripts
+[Object[]] String or ScriptBlocks to execute.
+
+.PARAMETER Variables
+[System.Collections.ArrayList] The variables to set in the new Pipeline.
+Each value is a hashtable that contains the parameters to use with
+Set-Variable;
+    Name: the name of the variable to set
+    Value: the value of the variable to set
+    Scope: the scope of the variable
+
+.PARAMETER Environment
+[System.Collections.IDictionary] A Dictionary of environment key/values to
+set in the new Pipeline.
+
+.PARAMETER Modules
+[System.Collections.IDictionary] A Dictionary of PowerShell modules to
+import into the new Pipeline. The key is the name of the module and the
+value is a base64 string of the module util code.
+
+.PARAMETER ModuleName
+[String] The name of the module that is being executed.
+
+.PARAMETER Breakpoints
+A list of line breakpoints to add to the runspace debugger. This is used to
+track module and module_utils coverage.
+#>
+param(
+    [Object[]]$Scripts,
+    [System.Collections.ArrayList][AllowEmptyCollection()]$Variables,
+    [System.Collections.IDictionary]$Environment,
+    [System.Collections.IDictionary]$Modules,
+    [String]$ModuleName,
+    [System.Management.Automation.LineBreakpoint[]]$Breakpoints = @()
+)
+
+Write-AnsibleLog "INFO - creating new PowerShell pipeline for $ModuleName" "module_wrapper"
+$ps = [PowerShell]::Create()
+
+# do not set ErrorActionPreference for script
+if ($ModuleName -ne "script") {
+    $ps.Runspace.SessionStateProxy.SetVariable("ErrorActionPreference", "Stop")
+}
+
+# force input encoding to preamble-free UTF8 so PS sub-processes (eg,
+# Start-Job) don't blow up. This is only required for WinRM, a PSRP
+# runspace doesn't have a host console and this will bomb out
+if ($host.Name -eq "ConsoleHost") {
+    Write-AnsibleLog "INFO - setting console input encoding to UTF8 for $ModuleName" "module_wrapper"
+    $ps.AddScript('[Console]::InputEncoding = New-Object Text.UTF8Encoding $false').AddStatement() > $null
+}
+
+# set the variables
+foreach ($variable in $Variables) {
+    Write-AnsibleLog "INFO - setting variable '$($variable.Name)' for $ModuleName" "module_wrapper"
+    $ps.AddCommand("Set-Variable").AddParameters($variable).AddStatement() > $null
+}
+
+# set the environment vars
+if ($Environment) {
+    # Escaping quotes can be problematic, instead just pass the string to the runspace and set it directly.
+    Write-AnsibleLog "INFO - setting environment vars for $ModuleName" "module_wrapper"
+    $ps.Runspace.SessionStateProxy.SetVariable("_AnsibleEnvironment", $Environment)
+    $ps.AddScript(@'
+foreach ($env_kv in $_AnsibleEnvironment.GetEnumerator()) {
+    [System.Environment]::SetEnvironmentVariable($env_kv.Key, $env_kv.Value)
+}
+'@).AddStatement() > $null
+}
+
+# import the PS modules
+if ($Modules) {
+    foreach ($module in $Modules.GetEnumerator()) {
+        Write-AnsibleLog "INFO - create module util '$($module.Key)' for $ModuleName" "module_wrapper"
+        $module_name = $module.Key
+        $module_code = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($module.Value))
+        $ps.AddCommand("New-Module").AddParameters(@{Name = $module_name; ScriptBlock = [ScriptBlock]::Create($module_code) }) > $null
+        $ps.AddCommand("Import-Module").AddParameter("WarningAction", "SilentlyContinue") > $null
+        $ps.AddCommand("Out-Null").AddStatement() > $null
+    }
+}
+
+# redefine Write-Host to dump to output instead of failing
+# lots of scripts still use it
+$ps.AddScript('Function Write-Host($msg) { Write-Output -InputObject $msg }').AddStatement() > $null
+
+# add the scripts and run
+foreach ($script in $Scripts) {
+    $ps.AddScript($script).AddStatement() > $null
+}
+
+if ($Breakpoints.Count -gt 0) {
+    Write-AnsibleLog "INFO - adding breakpoint to runspace that will run the modules" "module_wrapper"
+    if ($PSVersionTable.PSVersion.Major -eq 3) {
+        # The SetBreakpoints method was only added in PowerShell v4+. We need to rely on a private method to
+        # achieve the same functionality in this older PowerShell version. This should be removed once we drop
+        # support for PowerShell v3.
+        $set_method = $ps.Runspace.Debugger.GetType().GetMethod(
+            'AddLineBreakpoint', [System.Reflection.BindingFlags]'Instance, NonPublic'
+        )
+        foreach ($b in $Breakpoints) {
+            $set_method.Invoke($ps.Runspace.Debugger, [Object[]]@(, $b)) > $null
+        }
+    }
+    else {
+        $ps.Runspace.Debugger.SetBreakpoints($Breakpoints)
+    }
+}
+
+Write-AnsibleLog "INFO - start module exec with Invoke() - $ModuleName" "module_wrapper"
+
+# temporarily override the stdout stream and create our own in a StringBuilder
+# we use this to ensure there's always an Out pipe and that we capture the
+# output for things like async or psrp
+$orig_out = [System.Console]::Out
+$sb = New-Object -TypeName System.Text.StringBuilder
+$new_out = New-Object -TypeName System.IO.StringWriter -ArgumentList $sb
+try {
+    [System.Console]::SetOut($new_out)
+    $module_output = $ps.Invoke()
+}
+catch {
+    # uncaught exception while executing module, present a prettier error for
+    # Ansible to parse
+    $error_params = @{
+        Message = "Unhandled exception while executing module"
+        ErrorRecord = $_
+    }
+
+    # Be more defensive when trying to find the InnerException in case it isn't
+    # set. This shouldn't ever be the case but if it is then it makes it more
+    # difficult to track down the problem.
+    if ($_.Exception.PSObject.Properties.Name -contains "InnerException") {
+        $inner_exception = $_.Exception.InnerException
+        if ($inner_exception.PSObject.Properties.Name -contains "ErrorRecord") {
+            $error_params.ErrorRecord = $inner_exception.ErrorRecord
+        }
+    }
+
+    Write-AnsibleError @error_params
+    $host.SetShouldExit(1)
+    return
+}
+finally {
+    [System.Console]::SetOut($orig_out)
+    $new_out.Dispose()
+}
+
+# other types of errors may not throw an exception in Invoke but rather just
+# set the pipeline state to failed
+if ($ps.InvocationStateInfo.State -eq "Failed" -and $ModuleName -ne "script") {
+    $reason = $ps.InvocationStateInfo.Reason
+    $error_params = @{
+        Message = "Unhandled exception while executing module"
+    }
+
+    # The error record should always be set on the reason but this does not
+    # always happen on Server 2008 R2 for some reason (probably memory hotfix).
+    # Be defensive when trying to get the error record and fall back to other
+    # options.
+    if ($null -eq $reason) {
+        $error_params.Message += ": Unknown error"
+    }
+    elseif ($reason.PSObject.Properties.Name -contains "ErrorRecord") {
+        $error_params.ErrorRecord = $reason.ErrorRecord
+    }
+    else {
+        $error_params.Message += ": $($reason.ToString())"
+    }
+
+    Write-AnsibleError @error_params
+    $host.SetShouldExit(1)
+    return
+}
+
+Write-AnsibleLog "INFO - module exec ended $ModuleName" "module_wrapper"
+$stdout = $sb.ToString()
+if ($stdout) {
+    Write-Output -InputObject $stdout
+}
+if ($module_output.Count -gt 0) {
+    # do not output if empty collection
+    Write-AnsibleLog "INFO - using the output stream for module output - $ModuleName" "module_wrapper"
+    Write-Output -InputObject ($module_output -join "`r`n")
+}
+
+# we attempt to get the return code from the LASTEXITCODE variable
+# this is set explicitly in newer style variables when calling
+# ExitJson and FailJson. If set we set the current hosts' exit code
+# to that same value
+$rc = $ps.Runspace.SessionStateProxy.GetVariable("LASTEXITCODE")
+if ($null -ne $rc) {
+    Write-AnsibleLog "INFO - got an rc of $rc from $ModuleName exec" "module_wrapper"
+    $host.SetShouldExit($rc)
+}
+
+# PS3 doesn't properly set HadErrors in many cases, inspect the error stream as a fallback
+# with the trap handler that's now in place, this should only write to the output if
+# $ErrorActionPreference != "Stop", that's ok because this is sent to the stderr output
+# for a user to manually debug if something went horribly wrong
+if ($ps.HadErrors -or ($PSVersionTable.PSVersion.Major -lt 4 -and $ps.Streams.Error.Count -gt 0)) {
+    Write-AnsibleLog "WARN - module had errors, outputting error info $ModuleName" "module_wrapper"
+    # if the rc wasn't explicitly set, we return an exit code of 1
+    if ($null -eq $rc) {
+        $host.SetShouldExit(1)
+    }
+
+    # output each error to the error stream of the current pipeline
+    foreach ($err in $ps.Streams.Error) {
+        $error_msg = Format-AnsibleException -ErrorRecord $err
+
+        # need to use the current hosts's UI class as we may not have
+        # a console to write the stderr to, e.g. psrp
+        Write-AnsibleLog "WARN - error msg for for $($ModuleName):`r`n$error_msg" "module_wrapper"
+        $host.UI.WriteErrorLine($error_msg)
+    }
+}
diff --git a/lib/ansible/executor/process/__init__.py b/lib/ansible/executor/process/__init__.py
new file mode 100644
index 0000000..ae8ccff
--- /dev/null
+++ b/lib/ansible/executor/process/__init__.py
@@ -0,0 +1,20 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
diff --git a/lib/ansible/executor/process/worker.py b/lib/ansible/executor/process/worker.py
new file mode 100644
index 0000000..5113b83
--- /dev/null
+++ b/lib/ansible/executor/process/worker.py
@@ -0,0 +1,226 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+import sys
+import traceback
+
+from jinja2.exceptions import TemplateNotFound
+
+from ansible.errors import AnsibleConnectionFailure
+from ansible.executor.task_executor import TaskExecutor
+from ansible.module_utils._text import to_text
+from ansible.utils.display import Display
+from ansible.utils.multiprocessing import context as multiprocessing_context
+
+__all__ = ['WorkerProcess']
+
+display = Display()
+
+
+class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defined]
+    '''
+    The worker thread class, which uses TaskExecutor to run tasks
+    read from a job queue and pushes results into a results queue
+    for reading later.
+    '''
+
+    def __init__(self, final_q, task_vars, host, task, play_context, loader, variable_manager, shared_loader_obj):
+
+        super(WorkerProcess, self).__init__()
+        # takes a task queue manager as the sole param:
+        self._final_q = final_q
+        self._task_vars = task_vars
+        self._host = host
+        self._task = task
+        self._play_context = play_context
+        self._loader = loader
+        self._variable_manager = variable_manager
+        self._shared_loader_obj = shared_loader_obj
+
+        # NOTE: this works due to fork, if switching to threads this should change to per thread storage of temp files
+        # clear var to ensure we only delete files for this child
+        self._loader._tempfiles = set()
+
+    def _save_stdin(self):
+        self._new_stdin = None
+        try:
+            if sys.stdin.isatty() and sys.stdin.fileno() is not None:
+                try:
+                    self._new_stdin = os.fdopen(os.dup(sys.stdin.fileno()))
+                except OSError:
+                    # couldn't dupe stdin, most likely because it's
+                    # not a valid file descriptor
+                    pass
+        except (AttributeError, ValueError):
+            # couldn't get stdin's fileno
+            pass
+
+        if self._new_stdin is None:
+            self._new_stdin = open(os.devnull)
+
+    def start(self):
+        '''
+        multiprocessing.Process replaces the worker's stdin with a new file
+        but we wish to preserve it if it is connected to a terminal.
+        Therefore dup a copy prior to calling the real start(),
+        ensuring the descriptor is preserved somewhere in the new child, and
+        make sure it is closed in the parent when start() completes.
+        '''
+
+        self._save_stdin()
+        # FUTURE: this lock can be removed once a more generalized pre-fork thread pause is in place
+        with display._lock:
+            try:
+                return super(WorkerProcess, self).start()
+            finally:
+                self._new_stdin.close()
+
+    def _hard_exit(self, e):
+        '''
+        There is no safe exception to return to higher level code that does not
+        risk an innocent try/except finding itself executing in the wrong
+        process. All code executing above WorkerProcess.run() on the stack
+        conceptually belongs to another program.
+        '''
+
+        try:
+            display.debug(u"WORKER HARD EXIT: %s" % to_text(e))
+        except BaseException:
+            # If the cause of the fault is IOError being generated by stdio,
+            # attempting to log a debug message may trigger another IOError.
+            # Try printing once then give up.
+            pass
+
+        os._exit(1)
+
+    def run(self):
+        '''
+        Wrap _run() to ensure no possibility an errant exception can cause
+        control to return to the StrategyBase task loop, or any other code
+        higher in the stack.
+
+        As multiprocessing in Python 2.x provides no protection, it is possible
+        a try/except added in far-away code can cause a crashed child process
+        to suddenly assume the role and prior state of its parent.
+        '''
+        try:
+            return self._run()
+        except BaseException as e:
+            self._hard_exit(e)
+        finally:
+            # This is a hack, pure and simple, to work around a potential deadlock
+            # in ``multiprocessing.Process`` when flushing stdout/stderr during process
+            # shutdown.
+            #
+            # We should no longer have a problem with ``Display``, as it now proxies over
+            # the queue from a fork. However, to avoid any issues with plugins that may
+            # be doing their own printing, this has been kept.
+            #
+            # This happens at the very end to avoid that deadlock, by simply side
+            # stepping it. This should not be treated as a long term fix.
+            #
+            # TODO: Evaluate migrating away from the ``fork`` multiprocessing start method.
+            sys.stdout = sys.stderr = open(os.devnull, 'w')
+
+    def _run(self):
+        '''
+        Called when the process is started.  Pushes the result onto the
+        results queue. We also remove the host from the blocked hosts list, to
+        signify that they are ready for their next task.
+        '''
+
+        # import cProfile, pstats, StringIO
+        # pr = cProfile.Profile()
+        # pr.enable()
+
+        # Set the queue on Display so calls to Display.display are proxied over the queue
+        display.set_queue(self._final_q)
+
+        try:
+            # execute the task and build a TaskResult from the result
+            display.debug("running TaskExecutor() for %s/%s" % (self._host, self._task))
+            executor_result = TaskExecutor(
+                self._host,
+                self._task,
+                self._task_vars,
+                self._play_context,
+                self._new_stdin,
+                self._loader,
+                self._shared_loader_obj,
+                self._final_q
+            ).run()
+
+            display.debug("done running TaskExecutor() for %s/%s [%s]" % (self._host, self._task, self._task._uuid))
+            self._host.vars = dict()
+            self._host.groups = []
+
+            # put the result on the result queue
+            display.debug("sending task result for task %s" % self._task._uuid)
+            self._final_q.send_task_result(
+                self._host.name,
+                self._task._uuid,
+                executor_result,
+                task_fields=self._task.dump_attrs(),
+            )
+            display.debug("done sending task result for task %s" % self._task._uuid)
+
+        except AnsibleConnectionFailure:
+            self._host.vars = dict()
+            self._host.groups = []
+            self._final_q.send_task_result(
+                self._host.name,
+                self._task._uuid,
+                dict(unreachable=True),
+                task_fields=self._task.dump_attrs(),
+            )
+
+        except Exception as e:
+            if not isinstance(e, (IOError, EOFError, KeyboardInterrupt, SystemExit)) or isinstance(e, TemplateNotFound):
+                try:
+                    self._host.vars = dict()
+                    self._host.groups = []
+                    self._final_q.send_task_result(
+                        self._host.name,
+                        self._task._uuid,
+                        dict(failed=True, exception=to_text(traceback.format_exc()), stdout=''),
+                        task_fields=self._task.dump_attrs(),
+                    )
+                except Exception:
+                    display.debug(u"WORKER EXCEPTION: %s" % to_text(e))
+                    display.debug(u"WORKER TRACEBACK: %s" % to_text(traceback.format_exc()))
+                finally:
+                    self._clean_up()
+
+        display.debug("WORKER PROCESS EXITING")
+
+        # pr.disable()
+        # s = StringIO.StringIO()
+        # sortby = 'time'
+        # ps = pstats.Stats(pr, stream=s).sort_stats(sortby)
+        # ps.print_stats()
+        # with open('worker_%06d.stats' % os.getpid(), 'w') as f:
+        #     f.write(s.getvalue())
+
+    def _clean_up(self):
+        # NOTE: see note in init about forks
+        # ensure we cleanup all temp files for this worker
+        self._loader.cleanup_all_tmp_files()
diff --git a/lib/ansible/executor/stats.py b/lib/ansible/executor/stats.py
new file mode 100644
index 0000000..13a053b
--- /dev/null
+++ b/lib/ansible/executor/stats.py
@@ -0,0 +1,100 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from collections.abc import MutableMapping
+
+from ansible.utils.vars import merge_hash
+
+
+class AggregateStats:
+    ''' holds stats about per-host activity during playbook runs '''
+
+    def __init__(self):
+
+        self.processed = {}
+        self.failures = {}
+        self.ok = {}
+        self.dark = {}
+        self.changed = {}
+        self.skipped = {}
+        self.rescued = {}
+        self.ignored = {}
+
+        # user defined stats, which can be per host or global
+        self.custom = {}
+
+    def increment(self, what, host):
+        ''' helper function to bump a statistic '''
+
+        self.processed[host] = 1
+        prev = (getattr(self, what)).get(host, 0)
+        getattr(self, what)[host] = prev + 1
+
+    def decrement(self, what, host):
+        _what = getattr(self, what)
+        try:
+            if _what[host] - 1 < 0:
+                # This should never happen, but let's be safe
+                raise KeyError("Don't be so negative")
+            _what[host] -= 1
+        except KeyError:
+            _what[host] = 0
+
+    def summarize(self, host):
+        ''' return information about a particular host '''
+
+        return dict(
+            ok=self.ok.get(host, 0),
+            failures=self.failures.get(host, 0),
+            unreachable=self.dark.get(host, 0),
+            changed=self.changed.get(host, 0),
+            skipped=self.skipped.get(host, 0),
+            rescued=self.rescued.get(host, 0),
+            ignored=self.ignored.get(host, 0),
+        )
+
+    def set_custom_stats(self, which, what, host=None):
+        ''' allow setting of a custom stat'''
+
+        if host is None:
+            host = '_run'
+        if host not in self.custom:
+            self.custom[host] = {which: what}
+        else:
+            self.custom[host][which] = what
+
+    def update_custom_stats(self, which, what, host=None):
+        ''' allow aggregation of a custom stat'''
+
+        if host is None:
+            host = '_run'
+        if host not in self.custom or which not in self.custom[host]:
+            return self.set_custom_stats(which, what, host)
+
+        # mismatching types
+        if not isinstance(what, type(self.custom[host][which])):
+            return None
+
+        if isinstance(what, MutableMapping):
+            self.custom[host][which] = merge_hash(self.custom[host][which], what)
+        else:
+            # let overloaded + take care of other types
+            self.custom[host][which] += what
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
new file mode 100644
index 0000000..02ace8f
--- /dev/null
+++ b/lib/ansible/executor/task_executor.py
@@ -0,0 +1,1239 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+# (c) 2017 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+import pty
+import time
+import json
+import signal
+import subprocess
+import sys
+import termios
+import traceback
+
+from ansible import constants as C
+from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleConnectionFailure, AnsibleActionFail, AnsibleActionSkip
+from ansible.executor.task_result import TaskResult
+from ansible.executor.module_common import get_action_args_with_defaults
+from ansible.module_utils.parsing.convert_bool import boolean
+from ansible.module_utils.six import binary_type
+from ansible.module_utils._text import to_text, to_native
+from ansible.module_utils.connection import write_to_file_descriptor
+from ansible.playbook.conditional import Conditional
+from ansible.playbook.task import Task
+from ansible.plugins import get_plugin_class
+from ansible.plugins.loader import become_loader, cliconf_loader, connection_loader, httpapi_loader, netconf_loader, terminal_loader
+from ansible.template import Templar
+from ansible.utils.collection_loader import AnsibleCollectionConfig, AnsibleCollectionRef
+from ansible.utils.listify import listify_lookup_plugin_terms
+from ansible.utils.unsafe_proxy import to_unsafe_text, wrap_var
+from ansible.vars.clean import namespace_facts, clean_facts
+from ansible.utils.display import Display
+from ansible.utils.vars import combine_vars, isidentifier
+
+display = Display()
+
+
+RETURN_VARS = [x for x in C.MAGIC_VARIABLE_MAPPING.items() if 'become' not in x and '_pass' not in x]
+
+__all__ = ['TaskExecutor']
+
+
+class TaskTimeoutError(BaseException):
+    pass
+
+
+def task_timeout(signum, frame):
+    raise TaskTimeoutError
+
+
+def remove_omit(task_args, omit_token):
+    '''
+    Remove args with a value equal to the ``omit_token`` recursively
+    to align with now having suboptions in the argument_spec
+    '''
+
+    if not isinstance(task_args, dict):
+        return task_args
+
+    new_args = {}
+    for i in task_args.items():
+        if i[1] == omit_token:
+            continue
+        elif isinstance(i[1], dict):
+            new_args[i[0]] = remove_omit(i[1], omit_token)
+        elif isinstance(i[1], list):
+            new_args[i[0]] = [remove_omit(v, omit_token) for v in i[1]]
+        else:
+            new_args[i[0]] = i[1]
+
+    return new_args
+
+
+class TaskExecutor:
+
+    '''
+    This is the main worker class for the executor pipeline, which
+    handles loading an action plugin to actually dispatch the task to
+    a given host. This class roughly corresponds to the old Runner()
+    class.
+    '''
+
+    def __init__(self, host, task, job_vars, play_context, new_stdin, loader, shared_loader_obj, final_q):
+        self._host = host
+        self._task = task
+        self._job_vars = job_vars
+        self._play_context = play_context
+        self._new_stdin = new_stdin
+        self._loader = loader
+        self._shared_loader_obj = shared_loader_obj
+        self._connection = None
+        self._final_q = final_q
+        self._loop_eval_error = None
+
+        self._task.squash()
+
+    def run(self):
+        '''
+        The main executor entrypoint, where we determine if the specified
+        task requires looping and either runs the task with self._run_loop()
+        or self._execute(). After that, the returned results are parsed and
+        returned as a dict.
+        '''
+
+        display.debug("in run() - task %s" % self._task._uuid)
+
+        try:
+            try:
+                items = self._get_loop_items()
+            except AnsibleUndefinedVariable as e:
+                # save the error raised here for use later
+                items = None
+                self._loop_eval_error = e
+
+            if items is not None:
+                if len(items) > 0:
+                    item_results = self._run_loop(items)
+
+                    # create the overall result item
+                    res = dict(results=item_results)
+
+                    # loop through the item results and set the global changed/failed/skipped result flags based on any item.
+                    res['skipped'] = True
+                    for item in item_results:
+                        if 'changed' in item and item['changed'] and not res.get('changed'):
+                            res['changed'] = True
+                        if res['skipped'] and ('skipped' not in item or ('skipped' in item and not item['skipped'])):
+                            res['skipped'] = False
+                        if 'failed' in item and item['failed']:
+                            item_ignore = item.pop('_ansible_ignore_errors')
+                            if not res.get('failed'):
+                                res['failed'] = True
+                                res['msg'] = 'One or more items failed'
+                                self._task.ignore_errors = item_ignore
+                            elif self._task.ignore_errors and not item_ignore:
+                                self._task.ignore_errors = item_ignore
+
+                        # ensure to accumulate these
+                        for array in ['warnings', 'deprecations']:
+                            if array in item and item[array]:
+                                if array not in res:
+                                    res[array] = []
+                                if not isinstance(item[array], list):
+                                    item[array] = [item[array]]
+                                res[array] = res[array] + item[array]
+                                del item[array]
+
+                    if not res.get('failed', False):
+                        res['msg'] = 'All items completed'
+                    if res['skipped']:
+                        res['msg'] = 'All items skipped'
+                else:
+                    res = dict(changed=False, skipped=True, skipped_reason='No items in the list', results=[])
+            else:
+                display.debug("calling self._execute()")
+                res = self._execute()
+                display.debug("_execute() done")
+
+            # make sure changed is set in the result, if it's not present
+            if 'changed' not in res:
+                res['changed'] = False
+
+            def _clean_res(res, errors='surrogate_or_strict'):
+                if isinstance(res, binary_type):
+                    return to_unsafe_text(res, errors=errors)
+                elif isinstance(res, dict):
+                    for k in res:
+                        try:
+                            res[k] = _clean_res(res[k], errors=errors)
+                        except UnicodeError:
+                            if k == 'diff':
+                                # If this is a diff, substitute a replacement character if the value
+                                # is undecodable as utf8.  (Fix #21804)
+                                display.warning("We were unable to decode all characters in the module return data."
+                                                " Replaced some in an effort to return as much as possible")
+                                res[k] = _clean_res(res[k], errors='surrogate_then_replace')
+                            else:
+                                raise
+                elif isinstance(res, list):
+                    for idx, item in enumerate(res):
+                        res[idx] = _clean_res(item, errors=errors)
+                return res
+
+            display.debug("dumping result to json")
+            res = _clean_res(res)
+            display.debug("done dumping result, returning")
+            return res
+        except AnsibleError as e:
+            return dict(failed=True, msg=wrap_var(to_text(e, nonstring='simplerepr')), _ansible_no_log=self._play_context.no_log)
+        except Exception as e:
+            return dict(failed=True, msg=wrap_var('Unexpected failure during module execution: %s' % (to_native(e, nonstring='simplerepr'))),
+                        exception=to_text(traceback.format_exc()), stdout='', _ansible_no_log=self._play_context.no_log)
+        finally:
+            try:
+                self._connection.close()
+            except AttributeError:
+                pass
+            except Exception as e:
+                display.debug(u"error closing connection: %s" % to_text(e))
+
+    def _get_loop_items(self):
+        '''
+        Loads a lookup plugin to handle the with_* portion of a task (if specified),
+        and returns the items result.
+        '''
+
+        # get search path for this task to pass to lookup plugins
+        self._job_vars['ansible_search_path'] = self._task.get_search_path()
+
+        # ensure basedir is always in (dwim already searches here but we need to display it)
+        if self._loader.get_basedir() not in self._job_vars['ansible_search_path']:
+            self._job_vars['ansible_search_path'].append(self._loader.get_basedir())
+
+        templar = Templar(loader=self._loader, variables=self._job_vars)
+        items = None
+        loop_cache = self._job_vars.get('_ansible_loop_cache')
+        if loop_cache is not None:
+            # _ansible_loop_cache may be set in `get_vars` when calculating `delegate_to`
+            # to avoid reprocessing the loop
+            items = loop_cache
+        elif self._task.loop_with:
+            if self._task.loop_with in self._shared_loader_obj.lookup_loader:
+                fail = True
+                if self._task.loop_with == 'first_found':
+                    # first_found loops are special. If the item is undefined then we want to fall through to the next value rather than failing.
+                    fail = False
+
+                loop_terms = listify_lookup_plugin_terms(terms=self._task.loop, templar=templar, fail_on_undefined=fail, convert_bare=False)
+                if not fail:
+                    loop_terms = [t for t in loop_terms if not templar.is_template(t)]
+
+                # get lookup
+                mylookup = self._shared_loader_obj.lookup_loader.get(self._task.loop_with, loader=self._loader, templar=templar)
+
+                # give lookup task 'context' for subdir (mostly needed for first_found)
+                for subdir in ['template', 'var', 'file']:  # TODO: move this to constants?
+                    if subdir in self._task.action:
+                        break
+                setattr(mylookup, '_subdir', subdir + 's')
+
+                # run lookup
+                items = wrap_var(mylookup.run(terms=loop_terms, variables=self._job_vars, wantlist=True))
+            else:
+                raise AnsibleError("Unexpected failure in finding the lookup named '%s' in the available lookup plugins" % self._task.loop_with)
+
+        elif self._task.loop is not None:
+            items = templar.template(self._task.loop)
+            if not isinstance(items, list):
+                raise AnsibleError(
+                    "Invalid data passed to 'loop', it requires a list, got this instead: %s."
+                    " Hint: If you passed a list/dict of just one element,"
+                    " try adding wantlist=True to your lookup invocation or use q/query instead of lookup." % items
+                )
+
+        return items
+
+    def _run_loop(self, items):
+        '''
+        Runs the task with the loop items specified and collates the result
+        into an array named 'results' which is inserted into the final result
+        along with the item for which the loop ran.
+        '''
+        task_vars = self._job_vars
+        templar = Templar(loader=self._loader, variables=task_vars)
+
+        self._task.loop_control.post_validate(templar=templar)
+
+        loop_var = self._task.loop_control.loop_var
+        index_var = self._task.loop_control.index_var
+        loop_pause = self._task.loop_control.pause
+        extended = self._task.loop_control.extended
+        extended_allitems = self._task.loop_control.extended_allitems
+        # ensure we always have a label
+        label = self._task.loop_control.label or '{{' + loop_var + '}}'
+
+        if loop_var in task_vars:
+            display.warning(u"%s: The loop variable '%s' is already in use. "
+                            u"You should set the `loop_var` value in the `loop_control` option for the task"
+                            u" to something else to avoid variable collisions and unexpected behavior." % (self._task, loop_var))
+
+        ran_once = False
+        no_log = False
+        items_len = len(items)
+        results = []
+        for item_index, item in enumerate(items):
+            task_vars['ansible_loop_var'] = loop_var
+
+            task_vars[loop_var] = item
+            if index_var:
+                task_vars['ansible_index_var'] = index_var
+                task_vars[index_var] = item_index
+
+            if extended:
+                task_vars['ansible_loop'] = {
+                    'index': item_index + 1,
+                    'index0': item_index,
+                    'first': item_index == 0,
+                    'last': item_index + 1 == items_len,
+                    'length': items_len,
+                    'revindex': items_len - item_index,
+                    'revindex0': items_len - item_index - 1,
+                }
+                if extended_allitems:
+                    task_vars['ansible_loop']['allitems'] = items
+                try:
+                    task_vars['ansible_loop']['nextitem'] = items[item_index + 1]
+                except IndexError:
+                    pass
+                if item_index - 1 >= 0:
+                    task_vars['ansible_loop']['previtem'] = items[item_index - 1]
+
+            # Update template vars to reflect current loop iteration
+            templar.available_variables = task_vars
+
+            # pause between loop iterations
+            if loop_pause and ran_once:
+                time.sleep(loop_pause)
+            else:
+                ran_once = True
+
+            try:
+                tmp_task = self._task.copy(exclude_parent=True, exclude_tasks=True)
+                tmp_task._parent = self._task._parent
+                tmp_play_context = self._play_context.copy()
+            except AnsibleParserError as e:
+                results.append(dict(failed=True, msg=to_text(e)))
+                continue
+
+            # now we swap the internal task and play context with their copies,
+            # execute, and swap them back so we can do the next iteration cleanly
+            (self._task, tmp_task) = (tmp_task, self._task)
+            (self._play_context, tmp_play_context) = (tmp_play_context, self._play_context)
+            res = self._execute(variables=task_vars)
+            task_fields = self._task.dump_attrs()
+            (self._task, tmp_task) = (tmp_task, self._task)
+            (self._play_context, tmp_play_context) = (tmp_play_context, self._play_context)
+
+            # update 'general no_log' based on specific no_log
+            no_log = no_log or tmp_task.no_log
+
+            # now update the result with the item info, and append the result
+            # to the list of results
+            res[loop_var] = item
+            res['ansible_loop_var'] = loop_var
+            if index_var:
+                res[index_var] = item_index
+                res['ansible_index_var'] = index_var
+            if extended:
+                res['ansible_loop'] = task_vars['ansible_loop']
+
+            res['_ansible_item_result'] = True
+            res['_ansible_ignore_errors'] = task_fields.get('ignore_errors')
+
+            # gets templated here unlike rest of loop_control fields, depends on loop_var above
+            try:
+                res['_ansible_item_label'] = templar.template(label)
+            except AnsibleUndefinedVariable as e:
+                res.update({
+                    'failed': True,
+                    'msg': 'Failed to template loop_control.label: %s' % to_text(e)
+                })
+
+            tr = TaskResult(
+                self._host.name,
+                self._task._uuid,
+                res,
+                task_fields=task_fields,
+            )
+            if tr.is_failed() or tr.is_unreachable():
+                self._final_q.send_callback('v2_runner_item_on_failed', tr)
+            elif tr.is_skipped():
+                self._final_q.send_callback('v2_runner_item_on_skipped', tr)
+            else:
+                if getattr(self._task, 'diff', False):
+                    self._final_q.send_callback('v2_on_file_diff', tr)
+                if self._task.action not in C._ACTION_INVENTORY_TASKS:
+                    self._final_q.send_callback('v2_runner_item_on_ok', tr)
+
+            results.append(res)
+            del task_vars[loop_var]
+
+            # clear 'connection related' plugin variables for next iteration
+            if self._connection:
+                clear_plugins = {
+                    'connection': self._connection._load_name,
+                    'shell': self._connection._shell._load_name
+                }
+                if self._connection.become:
+                    clear_plugins['become'] = self._connection.become._load_name
+
+                for plugin_type, plugin_name in clear_plugins.items():
+                    for var in C.config.get_plugin_vars(plugin_type, plugin_name):
+                        if var in task_vars and var not in self._job_vars:
+                            del task_vars[var]
+
+        self._task.no_log = no_log
+
+        return results
+
+    def _execute(self, variables=None):
+        '''
+        The primary workhorse of the executor system, this runs the task
+        on the specified host (which may be the delegated_to host) and handles
+        the retry/until and block rescue/always execution
+        '''
+
+        if variables is None:
+            variables = self._job_vars
+
+        templar = Templar(loader=self._loader, variables=variables)
+
+        context_validation_error = None
+
+        # a certain subset of variables exist.
+        tempvars = variables.copy()
+
+        try:
+            # TODO: remove play_context as this does not take delegation nor loops correctly into account,
+            # the task itself should hold the correct values for connection/shell/become/terminal plugin options to finalize.
+            #  Kept for now for backwards compatibility and a few functions that are still exclusive to it.
+
+            # apply the given task's information to the connection info,
+            # which may override some fields already set by the play or
+            # the options specified on the command line
+            self._play_context = self._play_context.set_task_and_variable_override(task=self._task, variables=variables, templar=templar)
+
+            # fields set from the play/task may be based on variables, so we have to
+            # do the same kind of post validation step on it here before we use it.
+            self._play_context.post_validate(templar=templar)
+
+            # now that the play context is finalized, if the remote_addr is not set
+            # default to using the host's address field as the remote address
+            if not self._play_context.remote_addr:
+                self._play_context.remote_addr = self._host.address
+
+            # We also add "magic" variables back into the variables dict to make sure
+            self._play_context.update_vars(tempvars)
+
+        except AnsibleError as e:
+            # save the error, which we'll raise later if we don't end up
+            # skipping this task during the conditional evaluation step
+            context_validation_error = e
+
+        no_log = self._play_context.no_log
+
+        # Evaluate the conditional (if any) for this task, which we do before running
+        # the final task post-validation. We do this before the post validation due to
+        # the fact that the conditional may specify that the task be skipped due to a
+        # variable not being present which would otherwise cause validation to fail
+        try:
+            if not self._task.evaluate_conditional(templar, tempvars):
+                display.debug("when evaluation is False, skipping this task")
+                return dict(changed=False, skipped=True, skip_reason='Conditional result was False', _ansible_no_log=no_log)
+        except AnsibleError as e:
+            # loop error takes precedence
+            if self._loop_eval_error is not None:
+                # Display the error from the conditional as well to prevent
+                # losing information useful for debugging.
+                display.v(to_text(e))
+                raise self._loop_eval_error  # pylint: disable=raising-bad-type
+            raise
+
+        # Not skipping, if we had loop error raised earlier we need to raise it now to halt the execution of this task
+        if self._loop_eval_error is not None:
+            raise self._loop_eval_error  # pylint: disable=raising-bad-type
+
+        # if we ran into an error while setting up the PlayContext, raise it now, unless is known issue with delegation
+        # and undefined vars (correct values are in cvars later on and connection plugins, if still error, blows up there)
+        if context_validation_error is not None:
+            raiseit = True
+            if self._task.delegate_to:
+                if isinstance(context_validation_error, AnsibleUndefinedVariable):
+                    raiseit = False
+                elif isinstance(context_validation_error, AnsibleParserError):
+                    # parser error, might be cause by undef too
+                    orig_exc = getattr(context_validation_error, 'orig_exc', None)
+                    if isinstance(orig_exc, AnsibleUndefinedVariable):
+                        raiseit = False
+            if raiseit:
+                raise context_validation_error  # pylint: disable=raising-bad-type
+
+        # set templar to use temp variables until loop is evaluated
+        templar.available_variables = tempvars
+
+        # if this task is a TaskInclude, we just return now with a success code so the
+        # main thread can expand the task list for the given host
+        if self._task.action in C._ACTION_ALL_INCLUDE_TASKS:
+            include_args = self._task.args.copy()
+            include_file = include_args.pop('_raw_params', None)
+            if not include_file:
+                return dict(failed=True, msg="No include file was specified to the include")
+
+            include_file = templar.template(include_file)
+            return dict(include=include_file, include_args=include_args)
+
+        # if this task is a IncludeRole, we just return now with a success code so the main thread can expand the task list for the given host
+        elif self._task.action in C._ACTION_INCLUDE_ROLE:
+            include_args = self._task.args.copy()
+            return dict(include_args=include_args)
+
+        # Now we do final validation on the task, which sets all fields to their final values.
+        try:
+            self._task.post_validate(templar=templar)
+        except AnsibleError:
+            raise
+        except Exception:
+            return dict(changed=False, failed=True, _ansible_no_log=no_log, exception=to_text(traceback.format_exc()))
+        if '_variable_params' in self._task.args:
+            variable_params = self._task.args.pop('_variable_params')
+            if isinstance(variable_params, dict):
+                if C.INJECT_FACTS_AS_VARS:
+                    display.warning("Using a variable for a task's 'args' is unsafe in some situations "
+                                    "(see https://docs.ansible.com/ansible/devel/reference_appendices/faq.html#argsplat-unsafe)")
+                variable_params.update(self._task.args)
+                self._task.args = variable_params
+            else:
+                # if we didn't get a dict, it means there's garbage remaining after k=v parsing, just give up
+                # see https://github.com/ansible/ansible/issues/79862
+                raise AnsibleError(f"invalid or malformed argument: '{variable_params}'")
+
+        # update no_log to task value, now that we have it templated
+        no_log = self._task.no_log
+
+        # free tempvars up, not used anymore, cvars and vars_copy should be mainly used after this point
+        # updating the original 'variables' at the end
+        tempvars = {}
+
+        # setup cvars copy, used for all connection related templating
+        if self._task.delegate_to:
+            # use vars from delegated host (which already include task vars) instead of original host
+            cvars = variables.get('ansible_delegated_vars', {}).get(self._task.delegate_to, {})
+        else:
+            # just use normal host vars
+            cvars = variables
+
+        templar.available_variables = cvars
+
+        # use magic var if it exists, if not, let task inheritance do it's thing.
+        if cvars.get('ansible_connection') is not None:
+            current_connection = templar.template(cvars['ansible_connection'])
+        else:
+            current_connection = self._task.connection
+
+        # get the connection and the handler for this execution
+        if (not self._connection or
+                not getattr(self._connection, 'connected', False) or
+                not self._connection.matches_name([current_connection]) or
+                # pc compare, left here for old plugins, but should be irrelevant for those
+                # using get_option, since they are cleared each iteration.
+                self._play_context.remote_addr != self._connection._play_context.remote_addr):
+            self._connection = self._get_connection(cvars, templar, current_connection)
+        else:
+            # if connection is reused, its _play_context is no longer valid and needs
+            # to be replaced with the one templated above, in case other data changed
+            self._connection._play_context = self._play_context
+            self._set_become_plugin(cvars, templar, self._connection)
+
+        plugin_vars = self._set_connection_options(cvars, templar)
+
+        # make a copy of the job vars here, as we update them here and later,
+        # but don't want to pollute original
+        vars_copy = variables.copy()
+        # update with connection info (i.e ansible_host/ansible_user)
+        self._connection.update_vars(vars_copy)
+        templar.available_variables = vars_copy
+
+        # TODO: eventually remove as pc is taken out of the resolution path
+        # feed back into pc to ensure plugins not using get_option can get correct value
+        self._connection._play_context = self._play_context.set_task_and_variable_override(task=self._task, variables=vars_copy, templar=templar)
+
+        # for persistent connections, initialize socket path and start connection manager
+        if any(((self._connection.supports_persistence and C.USE_PERSISTENT_CONNECTIONS), self._connection.force_persistence)):
+            self._play_context.timeout = self._connection.get_option('persistent_command_timeout')
+            display.vvvv('attempting to start connection', host=self._play_context.remote_addr)
+            display.vvvv('using connection plugin %s' % self._connection.transport, host=self._play_context.remote_addr)
+
+            options = self._connection.get_options()
+            socket_path = start_connection(self._play_context, options, self._task._uuid)
+            display.vvvv('local domain socket path is %s' % socket_path, host=self._play_context.remote_addr)
+            setattr(self._connection, '_socket_path', socket_path)
+
+        # TODO: eventually remove this block as this should be a 'consequence' of 'forced_local' modules
+        # special handling for python interpreter for network_os, default to ansible python unless overridden
+        if 'ansible_network_os' in cvars and 'ansible_python_interpreter' not in cvars:
+            # this also avoids 'python discovery'
+            cvars['ansible_python_interpreter'] = sys.executable
+
+        # get handler
+        self._handler, module_context = self._get_action_handler_with_module_context(connection=self._connection, templar=templar)
+
+        if module_context is not None:
+            module_defaults_fqcn = module_context.resolved_fqcn
+        else:
+            module_defaults_fqcn = self._task.resolved_action
+
+        # Apply default params for action/module, if present
+        self._task.args = get_action_args_with_defaults(
+            module_defaults_fqcn, self._task.args, self._task.module_defaults, templar,
+            action_groups=self._task._parent._play._action_groups
+        )
+
+        # And filter out any fields which were set to default(omit), and got the omit token value
+        omit_token = variables.get('omit')
+        if omit_token is not None:
+            self._task.args = remove_omit(self._task.args, omit_token)
+
+        # Read some values from the task, so that we can modify them if need be
+        if self._task.until:
+            retries = self._task.retries
+            if retries is None:
+                retries = 3
+            elif retries <= 0:
+                retries = 1
+            else:
+                retries += 1
+        else:
+            retries = 1
+
+        delay = self._task.delay
+        if delay < 0:
+            delay = 1
+
+        display.debug("starting attempt loop")
+        result = None
+        for attempt in range(1, retries + 1):
+            display.debug("running the handler")
+            try:
+                if self._task.timeout:
+                    old_sig = signal.signal(signal.SIGALRM, task_timeout)
+                    signal.alarm(self._task.timeout)
+                result = self._handler.run(task_vars=vars_copy)
+            except (AnsibleActionFail, AnsibleActionSkip) as e:
+                return e.result
+            except AnsibleConnectionFailure as e:
+                return dict(unreachable=True, msg=to_text(e))
+            except TaskTimeoutError as e:
+                msg = 'The %s action failed to execute in the expected time frame (%d) and was terminated' % (self._task.action, self._task.timeout)
+                return dict(failed=True, msg=msg)
+            finally:
+                if self._task.timeout:
+                    signal.alarm(0)
+                    old_sig = signal.signal(signal.SIGALRM, old_sig)
+                self._handler.cleanup()
+            display.debug("handler run complete")
+
+            # preserve no log
+            result["_ansible_no_log"] = no_log
+
+            if self._task.action not in C._ACTION_WITH_CLEAN_FACTS:
+                result = wrap_var(result)
+
+            # update the local copy of vars with the registered value, if specified,
+            # or any facts which may have been generated by the module execution
+            if self._task.register:
+                if not isidentifier(self._task.register):
+                    raise AnsibleError("Invalid variable name in 'register' specified: '%s'" % self._task.register)
+
+                vars_copy[self._task.register] = result
+
+            if self._task.async_val > 0:
+                if self._task.poll > 0 and not result.get('skipped') and not result.get('failed'):
+                    result = self._poll_async_result(result=result, templar=templar, task_vars=vars_copy)
+                    if result.get('failed'):
+                        self._final_q.send_callback(
+                            'v2_runner_on_async_failed',
+                            TaskResult(self._host.name,
+                                       self._task._uuid,
+                                       result,
+                                       task_fields=self._task.dump_attrs()))
+                    else:
+                        self._final_q.send_callback(
+                            'v2_runner_on_async_ok',
+                            TaskResult(self._host.name,
+                                       self._task._uuid,
+                                       result,
+                                       task_fields=self._task.dump_attrs()))
+
+                # ensure no log is preserved
+                result["_ansible_no_log"] = no_log
+
+            # helper methods for use below in evaluating changed/failed_when
+            def _evaluate_changed_when_result(result):
+                if self._task.changed_when is not None and self._task.changed_when:
+                    cond = Conditional(loader=self._loader)
+                    cond.when = self._task.changed_when
+                    result['changed'] = cond.evaluate_conditional(templar, vars_copy)
+
+            def _evaluate_failed_when_result(result):
+                if self._task.failed_when:
+                    cond = Conditional(loader=self._loader)
+                    cond.when = self._task.failed_when
+                    failed_when_result = cond.evaluate_conditional(templar, vars_copy)
+                    result['failed_when_result'] = result['failed'] = failed_when_result
+                else:
+                    failed_when_result = False
+                return failed_when_result
+
+            if 'ansible_facts' in result and self._task.action not in C._ACTION_DEBUG:
+                if self._task.action in C._ACTION_WITH_CLEAN_FACTS:
+                    if self._task.delegate_to and self._task.delegate_facts:
+                        if '_ansible_delegated_vars' in vars_copy:
+                            vars_copy['_ansible_delegated_vars'].update(result['ansible_facts'])
+                        else:
+                            vars_copy['_ansible_delegated_vars'] = result['ansible_facts']
+                    else:
+                        vars_copy.update(result['ansible_facts'])
+                else:
+                    # TODO: cleaning of facts should eventually become part of taskresults instead of vars
+                    af = wrap_var(result['ansible_facts'])
+                    vars_copy['ansible_facts'] = combine_vars(vars_copy.get('ansible_facts', {}), namespace_facts(af))
+                    if C.INJECT_FACTS_AS_VARS:
+                        vars_copy.update(clean_facts(af))
+
+            # set the failed property if it was missing.
+            if 'failed' not in result:
+                # rc is here for backwards compatibility and modules that use it instead of 'failed'
+                if 'rc' in result and result['rc'] not in [0, "0"]:
+                    result['failed'] = True
+                else:
+                    result['failed'] = False
+
+            # Make attempts and retries available early to allow their use in changed/failed_when
+            if self._task.until:
+                result['attempts'] = attempt
+
+            # set the changed property if it was missing.
+            if 'changed' not in result:
+                result['changed'] = False
+
+            if self._task.action not in C._ACTION_WITH_CLEAN_FACTS:
+                result = wrap_var(result)
+
+            # re-update the local copy of vars with the registered value, if specified,
+            # or any facts which may have been generated by the module execution
+            # This gives changed/failed_when access to additional recently modified
+            # attributes of result
+            if self._task.register:
+                vars_copy[self._task.register] = result
+
+            # if we didn't skip this task, use the helpers to evaluate the changed/
+            # failed_when properties
+            if 'skipped' not in result:
+                condname = 'changed'
+
+                try:
+                    _evaluate_changed_when_result(result)
+                    condname = 'failed'
+                    _evaluate_failed_when_result(result)
+                except AnsibleError as e:
+                    result['failed'] = True
+                    result['%s_when_result' % condname] = to_text(e)
+
+            if retries > 1:
+                cond = Conditional(loader=self._loader)
+                cond.when = self._task.until
+                if cond.evaluate_conditional(templar, vars_copy):
+                    break
+                else:
+                    # no conditional check, or it failed, so sleep for the specified time
+                    if attempt < retries:
+                        result['_ansible_retry'] = True
+                        result['retries'] = retries
+                        display.debug('Retrying task, attempt %d of %d' % (attempt, retries))
+                        self._final_q.send_callback(
+                            'v2_runner_retry',
+                            TaskResult(
+                                self._host.name,
+                                self._task._uuid,
+                                result,
+                                task_fields=self._task.dump_attrs()
+                            )
+                        )
+                        time.sleep(delay)
+                        self._handler = self._get_action_handler(connection=self._connection, templar=templar)
+        else:
+            if retries > 1:
+                # we ran out of attempts, so mark the result as failed
+                result['attempts'] = retries - 1
+                result['failed'] = True
+
+        if self._task.action not in C._ACTION_WITH_CLEAN_FACTS:
+            result = wrap_var(result)
+
+        # do the final update of the local variables here, for both registered
+        # values and any facts which may have been created
+        if self._task.register:
+            variables[self._task.register] = result
+
+        if 'ansible_facts' in result and self._task.action not in C._ACTION_DEBUG:
+            if self._task.action in C._ACTION_WITH_CLEAN_FACTS:
+                variables.update(result['ansible_facts'])
+            else:
+                # TODO: cleaning of facts should eventually become part of taskresults instead of vars
+                af = wrap_var(result['ansible_facts'])
+                variables['ansible_facts'] = combine_vars(variables.get('ansible_facts', {}), namespace_facts(af))
+                if C.INJECT_FACTS_AS_VARS:
+                    variables.update(clean_facts(af))
+
+        # 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 is not None:
+            result['_ansible_notify'] = self._task.notify
+
+        # add the delegated vars to the result, so we can reference them
+        # on the results side without having to do any further templating
+        # also now add connection vars results when delegating
+        if self._task.delegate_to:
+            result["_ansible_delegated_vars"] = {'ansible_delegated_host': self._task.delegate_to}
+            for k in plugin_vars:
+                result["_ansible_delegated_vars"][k] = cvars.get(k)
+
+            # note: here for callbacks that rely on this info to display delegation
+            for requireshed in ('ansible_host', 'ansible_port', 'ansible_user', 'ansible_connection'):
+                if requireshed not in result["_ansible_delegated_vars"] and requireshed in cvars:
+                    result["_ansible_delegated_vars"][requireshed] = cvars.get(requireshed)
+
+        # and return
+        display.debug("attempt loop complete, returning result")
+        return result
+
+    def _poll_async_result(self, result, templar, task_vars=None):
+        '''
+        Polls for the specified JID to be complete
+        '''
+
+        if task_vars is None:
+            task_vars = self._job_vars
+
+        async_jid = result.get('ansible_job_id')
+        if async_jid is None:
+            return dict(failed=True, msg="No job id was returned by the async task")
+
+        # Create a new pseudo-task to run the async_status module, and run
+        # that (with a sleep for "poll" seconds between each retry) until the
+        # async time limit is exceeded.
+
+        async_task = Task.load(dict(action='async_status', args={'jid': async_jid}, environment=self._task.environment))
+
+        # FIXME: this is no longer the case, normal takes care of all, see if this can just be generalized
+        # Because this is an async task, the action handler is async. However,
+        # we need the 'normal' action handler for the status check, so get it
+        # now via the action_loader
+        async_handler = self._shared_loader_obj.action_loader.get(
+            'ansible.legacy.async_status',
+            task=async_task,
+            connection=self._connection,
+            play_context=self._play_context,
+            loader=self._loader,
+            templar=templar,
+            shared_loader_obj=self._shared_loader_obj,
+        )
+
+        time_left = self._task.async_val
+        while time_left > 0:
+            time.sleep(self._task.poll)
+
+            try:
+                async_result = async_handler.run(task_vars=task_vars)
+                # We do not bail out of the loop in cases where the failure
+                # is associated with a parsing error. The async_runner can
+                # have issues which result in a half-written/unparseable result
+                # file on disk, which manifests to the user as a timeout happening
+                # before it's time to timeout.
+                if (int(async_result.get('finished', 0)) == 1 or
+                        ('failed' in async_result and async_result.get('_ansible_parsed', False)) or
+                        'skipped' in async_result):
+                    break
+            except Exception as e:
+                # Connections can raise exceptions during polling (eg, network bounce, reboot); these should be non-fatal.
+                # On an exception, call the connection's reset method if it has one
+                # (eg, drop/recreate WinRM connection; some reused connections are in a broken state)
+                display.vvvv("Exception during async poll, retrying... (%s)" % to_text(e))
+                display.debug("Async poll exception was:\n%s" % to_text(traceback.format_exc()))
+                try:
+                    async_handler._connection.reset()
+                except AttributeError:
+                    pass
+
+                # Little hack to raise the exception if we've exhausted the timeout period
+                time_left -= self._task.poll
+                if time_left <= 0:
+                    raise
+            else:
+                time_left -= self._task.poll
+                self._final_q.send_callback(
+                    'v2_runner_on_async_poll',
+                    TaskResult(
+                        self._host.name,
+                        async_task._uuid,
+                        async_result,
+                        task_fields=async_task.dump_attrs(),
+                    ),
+                )
+
+        if int(async_result.get('finished', 0)) != 1:
+            if async_result.get('_ansible_parsed'):
+                return dict(failed=True, msg="async task did not complete within the requested time - %ss" % self._task.async_val, async_result=async_result)
+            else:
+                return dict(failed=True, msg="async task produced unparseable results", async_result=async_result)
+        else:
+            # If the async task finished, automatically cleanup the temporary
+            # status file left behind.
+            cleanup_task = Task.load(
+                {
+                    'async_status': {
+                        'jid': async_jid,
+                        'mode': 'cleanup',
+                    },
+                    'environment': self._task.environment,
+                }
+            )
+            cleanup_handler = self._shared_loader_obj.action_loader.get(
+                'ansible.legacy.async_status',
+                task=cleanup_task,
+                connection=self._connection,
+                play_context=self._play_context,
+                loader=self._loader,
+                templar=templar,
+                shared_loader_obj=self._shared_loader_obj,
+            )
+            cleanup_handler.run(task_vars=task_vars)
+            cleanup_handler.cleanup(force=True)
+            async_handler.cleanup(force=True)
+            return async_result
+
+    def _get_become(self, name):
+        become = become_loader.get(name)
+        if not become:
+            raise AnsibleError("Invalid become method specified, could not find matching plugin: '%s'. "
+                               "Use `ansible-doc -t become -l` to list available plugins." % name)
+        return become
+
+    def _get_connection(self, cvars, templar, current_connection):
+        '''
+        Reads the connection property for the host, and returns the
+        correct connection object from the list of connection plugins
+        '''
+
+        self._play_context.connection = current_connection
+
+        # TODO: play context has logic to update the connection for 'smart'
+        # (default value, will chose between ssh and paramiko) and 'persistent'
+        # (really paramiko), eventually this should move to task object itself.
+        conn_type = self._play_context.connection
+
+        connection, plugin_load_context = self._shared_loader_obj.connection_loader.get_with_context(
+            conn_type,
+            self._play_context,
+            self._new_stdin,
+            task_uuid=self._task._uuid,
+            ansible_playbook_pid=to_text(os.getppid())
+        )
+
+        if not connection:
+            raise AnsibleError("the connection plugin '%s' was not found" % conn_type)
+
+        self._set_become_plugin(cvars, templar, connection)
+
+        # Also backwards compat call for those still using play_context
+        self._play_context.set_attributes_from_plugin(connection)
+
+        return connection
+
+    def _set_become_plugin(self, cvars, templar, connection):
+        # load become plugin if needed
+        if cvars.get('ansible_become') is not None:
+            become = boolean(templar.template(cvars['ansible_become']))
+        else:
+            become = self._task.become
+
+        if become:
+            if cvars.get('ansible_become_method'):
+                become_plugin = self._get_become(templar.template(cvars['ansible_become_method']))
+            else:
+                become_plugin = self._get_become(self._task.become_method)
+
+        else:
+            # If become is not enabled on the task it needs to be removed from the connection plugin
+            # https://github.com/ansible/ansible/issues/78425
+            become_plugin = None
+
+        try:
+            connection.set_become_plugin(become_plugin)
+        except AttributeError:
+            # Older connection plugin that does not support set_become_plugin
+            pass
+
+        if become_plugin:
+            if getattr(connection.become, 'require_tty', False) and not getattr(connection, 'has_tty', False):
+                raise AnsibleError(
+                    "The '%s' connection does not provide a TTY which is required for the selected "
+                    "become plugin: %s." % (connection._load_name, become_plugin.name)
+                )
+
+            # Backwards compat for connection plugins that don't support become plugins
+            # Just do this unconditionally for now, we could move it inside of the
+            # AttributeError above later
+            self._play_context.set_become_plugin(become_plugin.name)
+
+    def _set_plugin_options(self, plugin_type, variables, templar, task_keys):
+        try:
+            plugin = getattr(self._connection, '_%s' % plugin_type)
+        except AttributeError:
+            # Some plugins are assigned to private attrs, ``become`` is not
+            plugin = getattr(self._connection, plugin_type)
+
+        # network_cli's "real" connection plugin is not named connection
+        # to avoid the confusion of having connection.connection
+        if plugin_type == "ssh_type_conn":
+            plugin_type = "connection"
+        option_vars = C.config.get_plugin_vars(plugin_type, plugin._load_name)
+        options = {}
+        for k in option_vars:
+            if k in variables:
+                options[k] = templar.template(variables[k])
+        # TODO move to task method?
+        plugin.set_options(task_keys=task_keys, var_options=options)
+
+        return option_vars
+
+    def _set_connection_options(self, variables, templar):
+
+        # keep list of variable names possibly consumed
+        varnames = []
+
+        # grab list of usable vars for this plugin
+        option_vars = C.config.get_plugin_vars('connection', self._connection._load_name)
+        varnames.extend(option_vars)
+
+        # create dict of 'templated vars'
+        options = {'_extras': {}}
+        for k in option_vars:
+            if k in variables:
+                options[k] = templar.template(variables[k])
+
+        # add extras if plugin supports them
+        if getattr(self._connection, 'allow_extras', False):
+            for k in variables:
+                if k.startswith('ansible_%s_' % self._connection._load_name) and k not in options:
+                    options['_extras'][k] = templar.template(variables[k])
+
+        task_keys = self._task.dump_attrs()
+
+        # The task_keys 'timeout' attr is the task's timeout, not the connection timeout.
+        # The connection timeout is threaded through the play_context for now.
+        task_keys['timeout'] = self._play_context.timeout
+
+        if self._play_context.password:
+            # The connection password is threaded through the play_context for
+            # now. This is something we ultimately want to avoid, but the first
+            # step is to get connection plugins pulling the password through the
+            # config system instead of directly accessing play_context.
+            task_keys['password'] = self._play_context.password
+
+        # Prevent task retries from overriding connection retries
+        del task_keys['retries']
+
+        # set options with 'templated vars' specific to this plugin and dependent ones
+        self._connection.set_options(task_keys=task_keys, var_options=options)
+        varnames.extend(self._set_plugin_options('shell', variables, templar, task_keys))
+
+        if self._connection.become is not None:
+            if self._play_context.become_pass:
+                # FIXME: eventually remove from task and play_context, here for backwards compat
+                # keep out of play objects to avoid accidental disclosure, only become plugin should have
+                # The become pass is already in the play_context if given on
+                # the CLI (-K). Make the plugin aware of it in this case.
+                task_keys['become_pass'] = self._play_context.become_pass
+
+            varnames.extend(self._set_plugin_options('become', variables, templar, task_keys))
+
+            # FOR BACKWARDS COMPAT:
+            for option in ('become_user', 'become_flags', 'become_exe', 'become_pass'):
+                try:
+                    setattr(self._play_context, option, self._connection.become.get_option(option))
+                except KeyError:
+                    pass  # some plugins don't support all base flags
+            self._play_context.prompt = self._connection.become.prompt
+
+        # deals with networking sub_plugins (network_cli/httpapi/netconf)
+        sub = getattr(self._connection, '_sub_plugin', None)
+        if sub is not None and sub.get('type') != 'external':
+            plugin_type = get_plugin_class(sub.get("obj"))
+            varnames.extend(self._set_plugin_options(plugin_type, variables, templar, task_keys))
+        sub_conn = getattr(self._connection, 'ssh_type_conn', None)
+        if sub_conn is not None:
+            varnames.extend(self._set_plugin_options("ssh_type_conn", variables, templar, task_keys))
+
+        return varnames
+
+    def _get_action_handler(self, connection, templar):
+        '''
+        Returns the correct action plugin to handle the requestion task action
+        '''
+        return self._get_action_handler_with_module_context(connection, templar)[0]
+
+    def _get_action_handler_with_module_context(self, connection, templar):
+        '''
+        Returns the correct action plugin to handle the requestion task action and the module context
+        '''
+        module_collection, separator, module_name = self._task.action.rpartition(".")
+        module_prefix = module_name.split('_')[0]
+        if module_collection:
+            # For network modules, which look for one action plugin per platform, look for the
+            # action plugin in the same collection as the module by prefixing the action plugin
+            # with the same collection.
+            network_action = "{0}.{1}".format(module_collection, module_prefix)
+        else:
+            network_action = module_prefix
+
+        collections = self._task.collections
+
+        # Check if the module has specified an action handler
+        module = self._shared_loader_obj.module_loader.find_plugin_with_context(
+            self._task.action, collection_list=collections
+        )
+        if not module.resolved or not module.action_plugin:
+            module = None
+        if module is not None:
+            handler_name = module.action_plugin
+        # let action plugin override module, fallback to 'normal' action plugin otherwise
+        elif self._shared_loader_obj.action_loader.has_plugin(self._task.action, collection_list=collections):
+            handler_name = self._task.action
+        elif all((module_prefix in C.NETWORK_GROUP_MODULES, self._shared_loader_obj.action_loader.has_plugin(network_action, collection_list=collections))):
+            handler_name = network_action
+            display.vvvv("Using network group action {handler} for {action}".format(handler=handler_name,
+                                                                                    action=self._task.action),
+                         host=self._play_context.remote_addr)
+        else:
+            # use ansible.legacy.normal to allow (historic) local action_plugins/ override without collections search
+            handler_name = 'ansible.legacy.normal'
+            collections = None  # until then, we don't want the task's collection list to be consulted; use the builtin
+
+        handler = self._shared_loader_obj.action_loader.get(
+            handler_name,
+            task=self._task,
+            connection=connection,
+            play_context=self._play_context,
+            loader=self._loader,
+            templar=templar,
+            shared_loader_obj=self._shared_loader_obj,
+            collection_list=collections
+        )
+
+        if not handler:
+            raise AnsibleError("the handler '%s' was not found" % handler_name)
+
+        return handler, module
+
+
+def start_connection(play_context, options, task_uuid):
+    '''
+    Starts the persistent connection
+    '''
+    candidate_paths = [C.ANSIBLE_CONNECTION_PATH or os.path.dirname(sys.argv[0])]
+    candidate_paths.extend(os.environ.get('PATH', '').split(os.pathsep))
+    for dirname in candidate_paths:
+        ansible_connection = os.path.join(dirname, 'ansible-connection')
+        if os.path.isfile(ansible_connection):
+            display.vvvv("Found ansible-connection at path {0}".format(ansible_connection))
+            break
+    else:
+        raise AnsibleError("Unable to find location of 'ansible-connection'. "
+                           "Please set or check the value of ANSIBLE_CONNECTION_PATH")
+
+    env = os.environ.copy()
+    env.update({
+        # HACK; most of these paths may change during the controller's lifetime
+        # (eg, due to late dynamic role includes, multi-playbook execution), without a way
+        # to invalidate/update, ansible-connection won't always see the same plugins the controller
+        # can.
+        'ANSIBLE_BECOME_PLUGINS': become_loader.print_paths(),
+        'ANSIBLE_CLICONF_PLUGINS': cliconf_loader.print_paths(),
+        'ANSIBLE_COLLECTIONS_PATH': to_native(os.pathsep.join(AnsibleCollectionConfig.collection_paths)),
+        'ANSIBLE_CONNECTION_PLUGINS': connection_loader.print_paths(),
+        'ANSIBLE_HTTPAPI_PLUGINS': httpapi_loader.print_paths(),
+        'ANSIBLE_NETCONF_PLUGINS': netconf_loader.print_paths(),
+        'ANSIBLE_TERMINAL_PLUGINS': terminal_loader.print_paths(),
+    })
+    verbosity = []
+    if display.verbosity:
+        verbosity.append('-%s' % ('v' * display.verbosity))
+    python = sys.executable
+    master, slave = pty.openpty()
+    p = subprocess.Popen(
+        [python, ansible_connection, *verbosity, to_text(os.getppid()), to_text(task_uuid)],
+        stdin=slave, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env
+    )
+    os.close(slave)
+
+    # We need to set the pty into noncanonical mode. This ensures that we
+    # can receive lines longer than 4095 characters (plus newline) without
+    # truncating.
+    old = termios.tcgetattr(master)
+    new = termios.tcgetattr(master)
+    new[3] = new[3] & ~termios.ICANON
+
+    try:
+        termios.tcsetattr(master, termios.TCSANOW, new)
+        write_to_file_descriptor(master, options)
+        write_to_file_descriptor(master, play_context.serialize())
+
+        (stdout, stderr) = p.communicate()
+    finally:
+        termios.tcsetattr(master, termios.TCSANOW, old)
+    os.close(master)
+
+    if p.returncode == 0:
+        result = json.loads(to_text(stdout, errors='surrogate_then_replace'))
+    else:
+        try:
+            result = json.loads(to_text(stderr, errors='surrogate_then_replace'))
+        except getattr(json.decoder, 'JSONDecodeError', ValueError):
+            # JSONDecodeError only available on Python 3.5+
+            result = {'error': to_text(stderr, errors='surrogate_then_replace')}
+
+    if 'messages' in result:
+        for level, message in result['messages']:
+            if level == 'log':
+                display.display(message, log_only=True)
+            elif level in ('debug', 'v', 'vv', 'vvv', 'vvvv', 'vvvvv', 'vvvvvv'):
+                getattr(display, level)(message, host=play_context.remote_addr)
+            else:
+                if hasattr(display, level):
+                    getattr(display, level)(message)
+                else:
+                    display.vvvv(message, host=play_context.remote_addr)
+
+    if 'error' in result:
+        if display.verbosity > 2:
+            if result.get('exception'):
+                msg = "The full traceback is:\n" + result['exception']
+                display.display(msg, color=C.COLOR_ERROR)
+        raise AnsibleError(result['error'])
+
+    return result['socket_path']
diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py
new file mode 100644
index 0000000..dcfc38a
--- /dev/null
+++ b/lib/ansible/executor/task_queue_manager.py
@@ -0,0 +1,456 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+import sys
+import tempfile
+import threading
+import time
+import multiprocessing.queues
+
+from ansible import constants as C
+from ansible import context
+from ansible.errors import AnsibleError
+from ansible.executor.play_iterator import PlayIterator
+from ansible.executor.stats import AggregateStats
+from ansible.executor.task_result import TaskResult
+from ansible.module_utils.six import string_types
+from ansible.module_utils._text import to_text, to_native
+from ansible.playbook.play_context import PlayContext
+from ansible.playbook.task import Task
+from ansible.plugins.loader import callback_loader, strategy_loader, module_loader
+from ansible.plugins.callback import CallbackBase
+from ansible.template import Templar
+from ansible.vars.hostvars import HostVars
+from ansible.vars.reserved import warn_if_reserved
+from ansible.utils.display import Display
+from ansible.utils.lock import lock_decorator
+from ansible.utils.multiprocessing import context as multiprocessing_context
+
+
+__all__ = ['TaskQueueManager']
+
+display = Display()
+
+
+class CallbackSend:
+    def __init__(self, method_name, *args, **kwargs):
+        self.method_name = method_name
+        self.args = args
+        self.kwargs = kwargs
+
+
+class DisplaySend:
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs
+
+
+class FinalQueue(multiprocessing.queues.Queue):
+    def __init__(self, *args, **kwargs):
+        kwargs['ctx'] = multiprocessing_context
+        super(FinalQueue, self).__init__(*args, **kwargs)
+
+    def send_callback(self, method_name, *args, **kwargs):
+        self.put(
+            CallbackSend(method_name, *args, **kwargs),
+            block=False
+        )
+
+    def send_task_result(self, *args, **kwargs):
+        if isinstance(args[0], TaskResult):
+            tr = args[0]
+        else:
+            tr = TaskResult(*args, **kwargs)
+        self.put(
+            tr,
+            block=False
+        )
+
+    def send_display(self, *args, **kwargs):
+        self.put(
+            DisplaySend(*args, **kwargs),
+            block=False
+        )
+
+
+class AnsibleEndPlay(Exception):
+    def __init__(self, result):
+        self.result = result
+
+
+class TaskQueueManager:
+
+    '''
+    This class handles the multiprocessing requirements of Ansible by
+    creating a pool of worker forks, a result handler fork, and a
+    manager object with shared datastructures/queues for coordinating
+    work between all processes.
+
+    The queue manager is responsible for loading the play strategy plugin,
+    which dispatches the Play's tasks to hosts.
+    '''
+
+    RUN_OK = 0
+    RUN_ERROR = 1
+    RUN_FAILED_HOSTS = 2
+    RUN_UNREACHABLE_HOSTS = 4
+    RUN_FAILED_BREAK_PLAY = 8
+    RUN_UNKNOWN_ERROR = 255
+
+    def __init__(self, inventory, variable_manager, loader, passwords, stdout_callback=None, run_additional_callbacks=True, run_tree=False, forks=None):
+
+        self._inventory = inventory
+        self._variable_manager = variable_manager
+        self._loader = loader
+        self._stats = AggregateStats()
+        self.passwords = passwords
+        self._stdout_callback = stdout_callback
+        self._run_additional_callbacks = run_additional_callbacks
+        self._run_tree = run_tree
+        self._forks = forks or 5
+
+        self._callbacks_loaded = False
+        self._callback_plugins = []
+        self._start_at_done = False
+
+        # make sure any module paths (if specified) are added to the module_loader
+        if context.CLIARGS.get('module_path', False):
+            for path in context.CLIARGS['module_path']:
+                if path:
+                    module_loader.add_directory(path)
+
+        # a special flag to help us exit cleanly
+        self._terminated = False
+
+        # dictionaries to keep track of failed/unreachable hosts
+        self._failed_hosts = dict()
+        self._unreachable_hosts = dict()
+
+        try:
+            self._final_q = FinalQueue()
+        except OSError as e:
+            raise AnsibleError("Unable to use multiprocessing, this is normally caused by lack of access to /dev/shm: %s" % to_native(e))
+
+        self._callback_lock = threading.Lock()
+
+        # A temporary file (opened pre-fork) used by connection
+        # plugins for inter-process locking.
+        self._connection_lockfile = tempfile.TemporaryFile()
+
+    def _initialize_processes(self, num):
+        self._workers = []
+
+        for i in range(num):
+            self._workers.append(None)
+
+    def load_callbacks(self):
+        '''
+        Loads all available callbacks, with the exception of those which
+        utilize the CALLBACK_TYPE option. When CALLBACK_TYPE is set to 'stdout',
+        only one such callback plugin will be loaded.
+        '''
+
+        if self._callbacks_loaded:
+            return
+
+        stdout_callback_loaded = False
+        if self._stdout_callback is None:
+            self._stdout_callback = C.DEFAULT_STDOUT_CALLBACK
+
+        if isinstance(self._stdout_callback, CallbackBase):
+            stdout_callback_loaded = True
+        elif isinstance(self._stdout_callback, string_types):
+            if self._stdout_callback not in callback_loader:
+                raise AnsibleError("Invalid callback for stdout specified: %s" % self._stdout_callback)
+            else:
+                self._stdout_callback = callback_loader.get(self._stdout_callback)
+                self._stdout_callback.set_options()
+                stdout_callback_loaded = True
+        else:
+            raise AnsibleError("callback must be an instance of CallbackBase or the name of a callback plugin")
+
+        # get all configured loadable callbacks (adjacent, builtin)
+        callback_list = list(callback_loader.all(class_only=True))
+
+        # add enabled callbacks that refer to collections, which might not appear in normal listing
+        for c in C.CALLBACKS_ENABLED:
+            # load all, as collection ones might be using short/redirected names and not a fqcn
+            plugin = callback_loader.get(c, class_only=True)
+
+            # TODO: check if this skip is redundant, loader should handle bad file/plugin cases already
+            if plugin:
+                # avoids incorrect and dupes possible due to collections
+                if plugin not in callback_list:
+                    callback_list.append(plugin)
+            else:
+                display.warning("Skipping callback plugin '%s', unable to load" % c)
+
+        # for each callback in the list see if we should add it to 'active callbacks' used in the play
+        for callback_plugin in callback_list:
+
+            callback_type = getattr(callback_plugin, 'CALLBACK_TYPE', '')
+            callback_needs_enabled = getattr(callback_plugin, 'CALLBACK_NEEDS_ENABLED', getattr(callback_plugin, 'CALLBACK_NEEDS_WHITELIST', False))
+
+            # try to get colleciotn world name first
+            cnames = getattr(callback_plugin, '_redirected_names', [])
+            if cnames:
+                # store the name the plugin was loaded as, as that's what we'll need to compare to the configured callback list later
+                callback_name = cnames[0]
+            else:
+                # fallback to 'old loader name'
+                (callback_name, _) = os.path.splitext(os.path.basename(callback_plugin._original_path))
+
+            display.vvvvv("Attempting to use '%s' callback." % (callback_name))
+            if callback_type == 'stdout':
+                # we only allow one callback of type 'stdout' to be loaded,
+                if callback_name != self._stdout_callback or stdout_callback_loaded:
+                    display.vv("Skipping callback '%s', as we already have a stdout callback." % (callback_name))
+                    continue
+                stdout_callback_loaded = True
+            elif callback_name == 'tree' and self._run_tree:
+                # TODO: remove special case for tree, which is an adhoc cli option --tree
+                pass
+            elif not self._run_additional_callbacks or (callback_needs_enabled and (
+                # only run if not adhoc, or adhoc was specifically configured to run + check enabled list
+                    C.CALLBACKS_ENABLED is None or callback_name not in C.CALLBACKS_ENABLED)):
+                # 2.x plugins shipped with ansible should require enabling, older or non shipped should load automatically
+                continue
+
+            try:
+                callback_obj = callback_plugin()
+                # avoid bad plugin not returning an object, only needed cause we do class_only load and bypass loader checks,
+                # really a bug in the plugin itself which we ignore as callback errors are not supposed to be fatal.
+                if callback_obj:
+                    # skip initializing if we already did the work for the same plugin (even with diff names)
+                    if callback_obj not in self._callback_plugins:
+                        callback_obj.set_options()
+                        self._callback_plugins.append(callback_obj)
+                    else:
+                        display.vv("Skipping callback '%s', already loaded as '%s'." % (callback_plugin, callback_name))
+                else:
+                    display.warning("Skipping callback '%s', as it does not create a valid plugin instance." % callback_name)
+                    continue
+            except Exception as e:
+                display.warning("Skipping callback '%s', unable to load due to: %s" % (callback_name, to_native(e)))
+                continue
+
+        self._callbacks_loaded = True
+
+    def run(self, play):
+        '''
+        Iterates over the roles/tasks in a play, using the given (or default)
+        strategy for queueing tasks. The default is the linear strategy, which
+        operates like classic Ansible by keeping all hosts in lock-step with
+        a given task (meaning no hosts move on to the next task until all hosts
+        are done with the current task).
+        '''
+
+        if not self._callbacks_loaded:
+            self.load_callbacks()
+
+        all_vars = self._variable_manager.get_vars(play=play)
+        templar = Templar(loader=self._loader, variables=all_vars)
+        warn_if_reserved(all_vars, templar.environment.globals.keys())
+
+        new_play = play.copy()
+        new_play.post_validate(templar)
+        new_play.handlers = new_play.compile_roles_handlers() + new_play.handlers
+
+        self.hostvars = HostVars(
+            inventory=self._inventory,
+            variable_manager=self._variable_manager,
+            loader=self._loader,
+        )
+
+        play_context = PlayContext(new_play, self.passwords, self._connection_lockfile.fileno())
+        if (self._stdout_callback and
+                hasattr(self._stdout_callback, 'set_play_context')):
+            self._stdout_callback.set_play_context(play_context)
+
+        for callback_plugin in self._callback_plugins:
+            if hasattr(callback_plugin, 'set_play_context'):
+                callback_plugin.set_play_context(play_context)
+
+        self.send_callback('v2_playbook_on_play_start', new_play)
+
+        # build the iterator
+        iterator = PlayIterator(
+            inventory=self._inventory,
+            play=new_play,
+            play_context=play_context,
+            variable_manager=self._variable_manager,
+            all_vars=all_vars,
+            start_at_done=self._start_at_done,
+        )
+
+        # adjust to # of workers to configured forks or size of batch, whatever is lower
+        self._initialize_processes(min(self._forks, iterator.batch_size))
+
+        # load the specified strategy (or the default linear one)
+        strategy = strategy_loader.get(new_play.strategy, self)
+        if strategy is None:
+            raise AnsibleError("Invalid play strategy specified: %s" % new_play.strategy, obj=play._ds)
+
+        # Because the TQM may survive multiple play runs, we start by marking
+        # any hosts as failed in the iterator here which may have been marked
+        # as failed in previous runs. Then we clear the internal list of failed
+        # hosts so we know what failed this round.
+        for host_name in self._failed_hosts.keys():
+            host = self._inventory.get_host(host_name)
+            iterator.mark_host_failed(host)
+        for host_name in self._unreachable_hosts.keys():
+            iterator._play._removed_hosts.append(host_name)
+
+        self.clear_failed_hosts()
+
+        # during initialization, the PlayContext will clear the start_at_task
+        # field to signal that a matching task was found, so check that here
+        # and remember it so we don't try to skip tasks on future plays
+        if context.CLIARGS.get('start_at_task') is not None and play_context.start_at_task is None:
+            self._start_at_done = True
+
+        # and run the play using the strategy and cleanup on way out
+        try:
+            play_return = strategy.run(iterator, play_context)
+        finally:
+            strategy.cleanup()
+            self._cleanup_processes()
+
+        # now re-save the hosts that failed from the iterator to our internal list
+        for host_name in iterator.get_failed_hosts():
+            self._failed_hosts[host_name] = True
+
+        if iterator.end_play:
+            raise AnsibleEndPlay(play_return)
+
+        return play_return
+
+    def cleanup(self):
+        display.debug("RUNNING CLEANUP")
+        self.terminate()
+        self._final_q.close()
+        self._cleanup_processes()
+        # We no longer flush on every write in ``Display.display``
+        # just ensure we've flushed during cleanup
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+    def _cleanup_processes(self):
+        if hasattr(self, '_workers'):
+            for attempts_remaining in range(C.WORKER_SHUTDOWN_POLL_COUNT - 1, -1, -1):
+                if not any(worker_prc and worker_prc.is_alive() for worker_prc in self._workers):
+                    break
+
+                if attempts_remaining:
+                    time.sleep(C.WORKER_SHUTDOWN_POLL_DELAY)
+                else:
+                    display.warning('One or more worker processes are still running and will be terminated.')
+
+            for worker_prc in self._workers:
+                if worker_prc and worker_prc.is_alive():
+                    try:
+                        worker_prc.terminate()
+                    except AttributeError:
+                        pass
+
+    def clear_failed_hosts(self):
+        self._failed_hosts = dict()
+
+    def get_inventory(self):
+        return self._inventory
+
+    def get_variable_manager(self):
+        return self._variable_manager
+
+    def get_loader(self):
+        return self._loader
+
+    def get_workers(self):
+        return self._workers[:]
+
+    def terminate(self):
+        self._terminated = True
+
+    def has_dead_workers(self):
+
+        # [<WorkerProcess(WorkerProcess-2, stopped[SIGKILL])>,
+        # <WorkerProcess(WorkerProcess-2, stopped[SIGTERM])>
+
+        defunct = False
+        for x in self._workers:
+            if getattr(x, 'exitcode', None):
+                defunct = True
+        return defunct
+
+    @lock_decorator(attr='_callback_lock')
+    def send_callback(self, method_name, *args, **kwargs):
+        for callback_plugin in [self._stdout_callback] + self._callback_plugins:
+            # a plugin that set self.disabled to True will not be called
+            # see osx_say.py example for such a plugin
+            if getattr(callback_plugin, 'disabled', False):
+                continue
+
+            # a plugin can opt in to implicit tasks (such as meta). It does this
+            # by declaring self.wants_implicit_tasks = True.
+            wants_implicit_tasks = getattr(callback_plugin, 'wants_implicit_tasks', False)
+
+            # try to find v2 method, fallback to v1 method, ignore callback if no method found
+            methods = []
+            for possible in [method_name, 'v2_on_any']:
+                gotit = getattr(callback_plugin, possible, None)
+                if gotit is None:
+                    gotit = getattr(callback_plugin, possible.removeprefix('v2_'), None)
+                if gotit is not None:
+                    methods.append(gotit)
+
+            # send clean copies
+            new_args = []
+
+            # If we end up being given an implicit task, we'll set this flag in
+            # the loop below. If the plugin doesn't care about those, then we
+            # check and continue to the next iteration of the outer loop.
+            is_implicit_task = False
+
+            for arg in args:
+                # FIXME: add play/task cleaners
+                if isinstance(arg, TaskResult):
+                    new_args.append(arg.clean_copy())
+                # elif isinstance(arg, Play):
+                # elif isinstance(arg, Task):
+                else:
+                    new_args.append(arg)
+
+                if isinstance(arg, Task) and arg.implicit:
+                    is_implicit_task = True
+
+            if is_implicit_task and not wants_implicit_tasks:
+                continue
+
+            for method in methods:
+                try:
+                    method(*new_args, **kwargs)
+                except Exception as e:
+                    # TODO: add config toggle to make this fatal or not?
+                    display.warning(u"Failure using method (%s) in callback plugin (%s): %s" % (to_text(method_name), to_text(callback_plugin), to_text(e)))
+                    from traceback import format_tb
+                    from sys import exc_info
+                    display.vvv('Callback Exception: \n' + ' '.join(format_tb(exc_info()[2])))
diff --git a/lib/ansible/executor/task_result.py b/lib/ansible/executor/task_result.py
new file mode 100644
index 0000000..543b860
--- /dev/null
+++ b/lib/ansible/executor/task_result.py
@@ -0,0 +1,154 @@
+# Copyright: (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from ansible import constants as C
+from ansible.parsing.dataloader import DataLoader
+from ansible.vars.clean import module_response_deepcopy, strip_internal_keys
+
+_IGNORE = ('failed', 'skipped')
+_PRESERVE = ('attempts', 'changed', 'retries')
+_SUB_PRESERVE = {'_ansible_delegated_vars': ('ansible_host', 'ansible_port', 'ansible_user', 'ansible_connection')}
+
+# stuff callbacks need
+CLEAN_EXCEPTIONS = (
+    '_ansible_verbose_always',  # for debug and other actions, to always expand data (pretty jsonification)
+    '_ansible_item_label',  # to know actual 'item' variable
+    '_ansible_no_log',  # jic we didnt clean up well enough, DON'T LOG
+    '_ansible_verbose_override',  # controls display of ansible_facts, gathering would be very noise with -v otherwise
+)
+
+
+class TaskResult:
+    '''
+    This class is responsible for interpreting the resulting data
+    from an executed task, and provides helper methods for determining
+    the result of a given task.
+    '''
+
+    def __init__(self, host, task, return_data, task_fields=None):
+        self._host = host
+        self._task = task
+
+        if isinstance(return_data, dict):
+            self._result = return_data.copy()
+        else:
+            self._result = DataLoader().load(return_data)
+
+        if task_fields is None:
+            self._task_fields = dict()
+        else:
+            self._task_fields = task_fields
+
+    @property
+    def task_name(self):
+        return self._task_fields.get('name', None) or self._task.get_name()
+
+    def is_changed(self):
+        return self._check_key('changed')
+
+    def is_skipped(self):
+        # loop results
+        if 'results' in self._result:
+            results = self._result['results']
+            # Loop tasks are only considered skipped if all items were skipped.
+            # some squashed results (eg, yum) are not dicts and can't be skipped individually
+            if results and all(isinstance(res, dict) and res.get('skipped', False) for res in results):
+                return True
+
+        # regular tasks and squashed non-dict results
+        return self._result.get('skipped', False)
+
+    def is_failed(self):
+        if 'failed_when_result' in self._result or \
+           'results' in self._result and True in [True for x in self._result['results'] if 'failed_when_result' in x]:
+            return self._check_key('failed_when_result')
+        else:
+            return self._check_key('failed')
+
+    def is_unreachable(self):
+        return self._check_key('unreachable')
+
+    def needs_debugger(self, globally_enabled=False):
+        _debugger = self._task_fields.get('debugger')
+        _ignore_errors = C.TASK_DEBUGGER_IGNORE_ERRORS and self._task_fields.get('ignore_errors')
+
+        ret = False
+        if globally_enabled and ((self.is_failed() and not _ignore_errors) or self.is_unreachable()):
+            ret = True
+
+        if _debugger in ('always',):
+            ret = True
+        elif _debugger in ('never',):
+            ret = False
+        elif _debugger in ('on_failed',) and self.is_failed() and not _ignore_errors:
+            ret = True
+        elif _debugger in ('on_unreachable',) and self.is_unreachable():
+            ret = True
+        elif _debugger in ('on_skipped',) and self.is_skipped():
+            ret = True
+
+        return ret
+
+    def _check_key(self, key):
+        '''get a specific key from the result or its items'''
+
+        if isinstance(self._result, dict) and key in self._result:
+            return self._result.get(key, False)
+        else:
+            flag = False
+            for res in self._result.get('results', []):
+                if isinstance(res, dict):
+                    flag |= res.get(key, False)
+            return flag
+
+    def clean_copy(self):
+
+        ''' returns 'clean' taskresult object '''
+
+        # FIXME: clean task_fields, _task and _host copies
+        result = TaskResult(self._host, self._task, {}, self._task_fields)
+
+        # statuses are already reflected on the event type
+        if result._task and result._task.action in C._ACTION_DEBUG:
+            # debug is verbose by default to display vars, no need to add invocation
+            ignore = _IGNORE + ('invocation',)
+        else:
+            ignore = _IGNORE
+
+        subset = {}
+        # preserve subset for later
+        for sub in _SUB_PRESERVE:
+            if sub in self._result:
+                subset[sub] = {}
+                for key in _SUB_PRESERVE[sub]:
+                    if key in self._result[sub]:
+                        subset[sub][key] = self._result[sub][key]
+
+        if isinstance(self._task.no_log, bool) and self._task.no_log or self._result.get('_ansible_no_log', False):
+            x = {"censored": "the output has been hidden due to the fact that 'no_log: true' was specified for this result"}
+
+            # preserve full
+            for preserve in _PRESERVE:
+                if preserve in self._result:
+                    x[preserve] = self._result[preserve]
+
+            result._result = x
+        elif self._result:
+            result._result = module_response_deepcopy(self._result)
+
+            # actualy remove
+            for remove_key in ignore:
+                if remove_key in result._result:
+                    del result._result[remove_key]
+
+            # remove almost ALL internal keys, keep ones relevant to callback
+            strip_internal_keys(result._result, exceptions=CLEAN_EXCEPTIONS)
+
+        # keep subset
+        result._result.update(subset)
+
+        return result