diff --git a/changelogs/fragments/action-plugin-always-cleanup.yml b/changelogs/fragments/action-plugin-always-cleanup.yml
new file mode 100644
index 00000000000000..15776974e1fe00
--- /dev/null
+++ b/changelogs/fragments/action-plugin-always-cleanup.yml
@@ -0,0 +1,7 @@
+bugfixes:
+- ActionBase - Add new ``cleanup`` method that is explicitly run by the
+  ``TaskExecutor`` to ensure that the shell plugins ``tmpdir`` is always
+  removed. This change means that individual action plugins need not be
+  responsible for removing the temporary directory, which ensures that
+  we don't have code paths that accidentally leave behind the temporary
+  directory.
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 0516479adc2107..eb32af7897a6a5 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -649,6 +649,8 @@ def _execute(self, variables=None):
                 return dict(failed=True, msg=to_text(e))
             except AnsibleConnectionFailure as e:
                 return dict(unreachable=True, msg=to_text(e))
+            finally:
+                self._handler.cleanup()
             display.debug("handler run complete")
 
             # preserve no log
@@ -853,6 +855,7 @@ def _poll_async_result(self, result, templar, task_vars=None):
             else:
                 return dict(failed=True, msg="async task produced unparseable results", async_result=async_result)
         else:
+            async_handler.cleanup(force=True)
             return async_result
 
     def _get_become(self, name):
diff --git a/lib/ansible/plugins/action/__init__.py b/lib/ansible/plugins/action/__init__.py
index 8b70d46cffaa2d..dc72e6c56c94fa 100644
--- a/lib/ansible/plugins/action/__init__.py
+++ b/lib/ansible/plugins/action/__init__.py
@@ -115,6 +115,18 @@ def run(self, tmp=None, task_vars=None):
 
         return result
 
+    def cleanup(self, force=False):
+        """Method to perform a clean up at the end of an action plugin execution
+
+        By default this is designed to clean up the shell tmpdir, and is toggled based on whether
+        async is in use
+
+        Action plugins may override this if they deem necessary, but should still call this method
+        via super
+        """
+        if force or not self._task.async_val:
+            self._remove_tmp_path(self._connection._shell.tmpdir)
+
     def get_plugin_option(self, plugin, option, default=None):
         """Helper to get an option from a plugin without having to use
         the try/except dance everywhere to set a default
diff --git a/test/integration/targets/listen_ports_facts/tasks/main.yml b/test/integration/targets/listen_ports_facts/tasks/main.yml
index 2ed69e00023828..02d6ed57ca8e62 100644
--- a/test/integration/targets/listen_ports_facts/tasks/main.yml
+++ b/test/integration/targets/listen_ports_facts/tasks/main.yml
@@ -77,3 +77,9 @@
   assert:
     that: 5555 in ansible_facts.udp_listen | map(attribute='port') | sort | list
   when: (ansible_os_family == "RedHat" and ansible_distribution_major_version|int >= 7) or ansible_os_family == "Debian"
+
+- name: kill all async commands
+  command: "kill -9 {{ item.pid }}"
+  loop: "{{ [tcp_listen, udp_listen]|flatten }}"
+  when: item.name == 'nc'
+  ignore_errors: true
diff --git a/test/integration/targets/remote_tmp/playbook.yml b/test/integration/targets/remote_tmp/playbook.yml
index 46a2846590c22f..cfd44b75e6e707 100644
--- a/test/integration/targets/remote_tmp/playbook.yml
+++ b/test/integration/targets/remote_tmp/playbook.yml
@@ -24,3 +24,32 @@
         - name: clean up test user
           user: name=tmptest state=absent
           become_user: root
+
+- name: Test tempdir is removed
+  hosts: testhost
+  gather_facts: false
+  tasks:
+    - file:
+        state: touch
+        path: "/{{ output_dir }}/65393"
+
+    - copy:
+        src: "/{{ output_dir }}/65393"
+        dest: "/{{ output_dir }}/65393.2"
+        remote_src: true
+
+    - find:
+        path: "~/.ansible/tmp"
+        use_regex: yes
+        patterns: 'AnsiballZ_.+\.py'
+        recurse: true
+      register: result
+
+    - debug:
+        var: result
+
+    - assert:
+        that:
+          # Should only be AnsiballZ_find.py because find is actively running
+          - result.files|length == 1
+          - result.files[0].path.endswith('/AnsiballZ_find.py')
diff --git a/test/integration/targets/remote_tmp/runme.sh b/test/integration/targets/remote_tmp/runme.sh
index 69efd6e01644c2..8d1eebd6b14623 100755
--- a/test/integration/targets/remote_tmp/runme.sh
+++ b/test/integration/targets/remote_tmp/runme.sh
@@ -2,4 +2,4 @@
 
 set -ux
 
-ansible-playbook -i ../../inventory playbook.yml -v "$@"
+ansible-playbook -i ../../inventory playbook.yml -e "output_dir=${OUTPUT_DIR}" -v "$@"