diff --git a/changelogs/fragments/62237-keep-unsafe-context.yml b/changelogs/fragments/62237-keep-unsafe-context.yml
new file mode 100644
index 00000000000..e4513e7e484
--- /dev/null
+++ b/changelogs/fragments/62237-keep-unsafe-context.yml
@@ -0,0 +1,5 @@
+bugfixes:
+- >
+  **security issue** - TaskExecutor - Ensure we don't erase unsafe context in TaskExecutor.run on bytes.
+  Only present in 2.9.0beta1
+  (https://github.com/ansible/ansible/issues/62237)
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 3f7c09a0a5f..0e741fb8ffd 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -28,7 +28,7 @@ from ansible.plugins.loader import become_loader, cliconf_loader, connection_loa
 from ansible.template import Templar
 from ansible.utils.collection_loader import AnsibleCollectionLoader
 from ansible.utils.listify import listify_lookup_plugin_terms
-from ansible.utils.unsafe_proxy import AnsibleUnsafe, wrap_var
+from ansible.utils.unsafe_proxy import AnsibleUnsafe, 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
@@ -152,7 +152,7 @@ class TaskExecutor:
 
             def _clean_res(res, errors='surrogate_or_strict'):
                 if isinstance(res, binary_type):
-                    return to_text(res, errors=errors)
+                    return to_unsafe_text(res, errors=errors)
                 elif isinstance(res, dict):
                     for k in res:
                         try:
diff --git a/lib/ansible/utils/unsafe_proxy.py b/lib/ansible/utils/unsafe_proxy.py
index 8b978c67d95..59c805d53e6 100644
--- a/lib/ansible/utils/unsafe_proxy.py
+++ b/lib/ansible/utils/unsafe_proxy.py
@@ -53,7 +53,7 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
-from ansible.module_utils._text import to_text
+from ansible.module_utils._text import to_bytes, to_text
 from ansible.module_utils.common._collections_compat import Mapping, MutableSequence, Set
 from ansible.module_utils.six import string_types, binary_type, text_type
 
@@ -126,3 +126,11 @@ def wrap_var(v):
         v = AnsibleUnsafeText(v)
 
     return v
+
+
+def to_unsafe_bytes(*args, **kwargs):
+    return wrap_var(to_bytes(*args, **kwargs))
+
+
+def to_unsafe_text(*args, **kwargs):
+    return wrap_var(to_text(*args, **kwargs))
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
index d7baa00d75e..dfbcebef6d0 100644
--- a/test/units/executor/test_task_executor.py
+++ b/test/units/executor/test_task_executor.py
@@ -27,6 +27,8 @@ from ansible.errors import AnsibleError
 from ansible.executor.task_executor import TaskExecutor, remove_omit
 from ansible.plugins.loader import action_loader, lookup_loader
 from ansible.parsing.yaml.objects import AnsibleUnicode
+from ansible.utils.unsafe_proxy import AnsibleUnsafeText, AnsibleUnsafeBytes
+from ansible.module_utils.six import text_type
 
 from units.mock.loader import DictDataLoader
 
@@ -101,6 +103,28 @@ class TestTaskExecutor(unittest.TestCase):
         res = te.run()
         self.assertIn("failed", res)
 
+    def test_task_executor_run_clean_res(self):
+        te = TaskExecutor(None, MagicMock(), None, None, None, None, None, None)
+        te._get_loop_items = MagicMock(return_value=[1])
+        te._run_loop = MagicMock(
+            return_value=[
+                {
+                    'unsafe_bytes': AnsibleUnsafeBytes(b'{{ $bar }}'),
+                    'unsafe_text': AnsibleUnsafeText(u'{{ $bar }}'),
+                    'bytes': b'bytes',
+                    'text': u'text',
+                    'int': 1,
+                }
+            ]
+        )
+        res = te.run()
+        data = res['results'][0]
+        self.assertIsInstance(data['unsafe_bytes'], AnsibleUnsafeText)
+        self.assertIsInstance(data['unsafe_text'], AnsibleUnsafeText)
+        self.assertIsInstance(data['bytes'], text_type)
+        self.assertIsInstance(data['text'], text_type)
+        self.assertIsInstance(data['int'], int)
+
     def test_task_executor_get_loop_items(self):
         fake_loader = DictDataLoader({})