diff --git a/changelogs/fragments/templates_types_datatagging.yml b/changelogs/fragments/templates_types_datatagging.yml
index 6a5e565bb71..ac65f6c668b 100644
--- a/changelogs/fragments/templates_types_datatagging.yml
+++ b/changelogs/fragments/templates_types_datatagging.yml
@@ -173,6 +173,9 @@ deprecated_features:
   - file loading - Loading text files with ``DataLoader`` containing data that cannot be decoded under the expected encoding is deprecated.
     In most cases the encoding must be UTF-8, although some plugins allow choosing a different encoding.
     Previously, invalid data was silently wrapped in Unicode surrogate escape sequences, often resulting in later errors or other data corruption.
+  - callback plugins - The v1 callback API (callback methods not prefixed with `v2_`) is deprecated.
+    Use `v2_` prefixed methods instead.
+  - callback plugins - The `v2_on_any` callback method is deprecated. Use specific callback methods instead.
 
 removed_features:
   - modules - Modules returning non-UTF8 strings now result in an error.
diff --git a/lib/ansible/_internal/_collection_proxy.py b/lib/ansible/_internal/_collection_proxy.py
new file mode 100644
index 00000000000..4e7321a5c16
--- /dev/null
+++ b/lib/ansible/_internal/_collection_proxy.py
@@ -0,0 +1,47 @@
+from __future__ import annotations as _annotations
+
+import collections.abc as _c
+import typing as _t
+
+_T_co = _t.TypeVar('_T_co', covariant=True)
+
+
+class SequenceProxy(_c.Sequence[_T_co]):
+    """A read-only sequence proxy."""
+
+    # DTFIX-RELEASE: needs unit test coverage
+
+    __slots__ = ('__value',)
+
+    def __init__(self, value: _c.Sequence[_T_co]) -> None:
+        self.__value = value
+
+    @_t.overload
+    def __getitem__(self, index: int) -> _T_co: ...
+
+    @_t.overload
+    def __getitem__(self, index: slice) -> _c.Sequence[_T_co]: ...
+
+    def __getitem__(self, index: int | slice) -> _T_co | _c.Sequence[_T_co]:
+        if isinstance(index, slice):
+            return self.__class__(self.__value[index])
+
+        return self.__value[index]
+
+    def __len__(self) -> int:
+        return len(self.__value)
+
+    def __contains__(self, item: object) -> bool:
+        return item in self.__value
+
+    def __iter__(self) -> _t.Iterator[_T_co]:
+        yield from self.__value
+
+    def __reversed__(self) -> _c.Iterator[_T_co]:
+        return reversed(self.__value)
+
+    def index(self, *args) -> int:
+        return self.__value.index(*args)
+
+    def count(self, value: object) -> int:
+        return self.__value.count(value)
diff --git a/lib/ansible/_internal/_json/__init__.py b/lib/ansible/_internal/_json/__init__.py
index 81cb409aeb9..5827ddc4cb8 100644
--- a/lib/ansible/_internal/_json/__init__.py
+++ b/lib/ansible/_internal/_json/__init__.py
@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import enum
 import json
 import typing as t
 
@@ -19,7 +20,9 @@ from ansible.module_utils._internal._datatag import (
 from ansible.module_utils._internal._json._profiles import _tagless
 from ansible.parsing.vault import EncryptedString
 from ansible._internal._datatag._tags import Origin, TrustedAsTemplate
+from ansible._internal._templating import _transform
 from ansible.module_utils import _internal
+from ansible.module_utils._internal import _datatag
 
 _T = t.TypeVar('_T')
 _sentinel = object()
@@ -52,6 +55,19 @@ class StateTrackingMixIn(HasCurrent):
         return self._stack[1:] + [self._current]
 
 
+class EncryptedStringBehavior(enum.Enum):
+    """How `AnsibleVariableVisitor` will handle instances of `EncryptedString`."""
+
+    PRESERVE = enum.auto()
+    """Preserves the unmodified `EncryptedString` instance."""
+    DECRYPT = enum.auto()
+    """Replaces the value with its decrypted plaintext."""
+    REDACT = enum.auto()
+    """Replaces the value with a placeholder string."""
+    FAIL = enum.auto()
+    """Raises an `AnsibleVariableTypeError` error."""
+
+
 class AnsibleVariableVisitor:
     """Utility visitor base class to recursively apply various behaviors and checks to variable object graphs."""
 
@@ -63,7 +79,9 @@ class AnsibleVariableVisitor:
         convert_mapping_to_dict: bool = False,
         convert_sequence_to_list: bool = False,
         convert_custom_scalars: bool = False,
-        allow_encrypted_string: bool = False,
+        convert_to_native_values: bool = False,
+        apply_transforms: bool = False,
+        encrypted_string_behavior: EncryptedStringBehavior = EncryptedStringBehavior.DECRYPT,
     ):
         super().__init__()  # supports StateTrackingMixIn
 
@@ -72,7 +90,16 @@ class AnsibleVariableVisitor:
         self.convert_mapping_to_dict = convert_mapping_to_dict
         self.convert_sequence_to_list = convert_sequence_to_list
         self.convert_custom_scalars = convert_custom_scalars
-        self.allow_encrypted_string = allow_encrypted_string
+        self.convert_to_native_values = convert_to_native_values
+        self.apply_transforms = apply_transforms
+        self.encrypted_string_behavior = encrypted_string_behavior
+
+        if apply_transforms:
+            from ansible._internal._templating import _engine
+
+            self._template_engine = _engine.TemplateEngine()
+        else:
+            self._template_engine = None
 
         self._current: t.Any = None  # supports StateTrackingMixIn
 
@@ -113,9 +140,19 @@ class AnsibleVariableVisitor:
 
         value_type = type(value)
 
+        if self.apply_transforms and value_type in _transform._type_transform_mapping:
+            value = self._template_engine.transform(value)
+            value_type = type(value)
+
+        # DTFIX-RELEASE: need to handle native copy for keys too
+        if self.convert_to_native_values and isinstance(value, _datatag.AnsibleTaggedObject):
+            value = value._native_copy()
+            value_type = type(value)
+
         result: _T
 
         # DTFIX-RELEASE: the visitor is ignoring dict/mapping keys except for debugging and schema-aware checking, it should be doing type checks on keys
+        #                keep in mind the allowed types for keys is a more restrictive set than for values (str and taggged str only, not EncryptedString)
         # DTFIX-RELEASE: some type lists being consulted (the ones from datatag) are probably too permissive, and perhaps should not be dynamic
 
         if (result := self._early_visit(value, value_type)) is not _sentinel:
@@ -127,8 +164,14 @@ class AnsibleVariableVisitor:
         elif value_type in _ANSIBLE_ALLOWED_NON_SCALAR_COLLECTION_VAR_TYPES:
             with self:  # supports StateTrackingMixIn
                 result = AnsibleTagHelper.tag_copy(value, (self._visit(k, v) for k, v in enumerate(t.cast(t.Iterable, value))), value_type=value_type)
-        elif self.allow_encrypted_string and isinstance(value, EncryptedString):
-            return value  # type: ignore[return-value]  # DTFIX-RELEASE: this should probably only be allowed for values in dict, not keys (set, dict)
+        elif self.encrypted_string_behavior != EncryptedStringBehavior.FAIL and isinstance(value, EncryptedString):
+            match self.encrypted_string_behavior:
+                case EncryptedStringBehavior.REDACT:
+                    result = "<redacted>"  # type: ignore[assignment]
+                case EncryptedStringBehavior.PRESERVE:
+                    result = value  # type: ignore[assignment]
+                case EncryptedStringBehavior.DECRYPT:
+                    result = str(value)  # type: ignore[assignment]
         elif self.convert_mapping_to_dict and _internal.is_intermediate_mapping(value):
             with self:  # supports StateTrackingMixIn
                 result = {k: self._visit(k, v) for k, v in value.items()}  # type: ignore[assignment]
diff --git a/lib/ansible/_internal/_json/_profiles/_legacy.py b/lib/ansible/_internal/_json/_profiles/_legacy.py
index 2b333e6da12..c1638468a9d 100644
--- a/lib/ansible/_internal/_json/_profiles/_legacy.py
+++ b/lib/ansible/_internal/_json/_profiles/_legacy.py
@@ -8,13 +8,12 @@ from __future__ import annotations as _annotations
 import datetime as _datetime
 import typing as _t
 
+from ansible._internal import _json
 from ansible._internal._datatag import _tags
 from ansible.module_utils._internal import _datatag
 from ansible.module_utils._internal._json import _profiles
 from ansible.parsing import vault as _vault
 
-from ... import _json
-
 
 class _Untrusted:
     """
@@ -48,7 +47,7 @@ class _LegacyVariableVisitor(_json.AnsibleVariableVisitor):
             convert_mapping_to_dict=convert_mapping_to_dict,
             convert_sequence_to_list=convert_sequence_to_list,
             convert_custom_scalars=convert_custom_scalars,
-            allow_encrypted_string=True,
+            encrypted_string_behavior=_json.EncryptedStringBehavior.PRESERVE,
         )
 
         self.invert_trust = invert_trust
diff --git a/lib/ansible/executor/process/worker.py b/lib/ansible/executor/process/worker.py
index ed7cafc5f57..2fb1aa15681 100644
--- a/lib/ansible/executor/process/worker.py
+++ b/lib/ansible/executor/process/worker.py
@@ -32,7 +32,7 @@ from ansible._internal import _task
 from ansible.errors import AnsibleConnectionFailure, AnsibleError
 from ansible.executor.task_executor import TaskExecutor
 from ansible.executor.task_queue_manager import FinalQueue, STDIN_FILENO, STDOUT_FILENO, STDERR_FILENO
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import _RawTaskResult
 from ansible.inventory.host import Host
 from ansible.module_utils.common.collections import is_sequence
 from ansible.module_utils.common.text.converters import to_text
@@ -227,7 +227,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
             init_plugin_loader(cli_collections_path)
 
         try:
-            # execute the task and build a TaskResult from the result
+            # execute the task and build a _RawTaskResult from the result
             display.debug("running TaskExecutor() for %s/%s" % (self._host, self._task))
             executor_result = TaskExecutor(
                 self._host,
@@ -257,7 +257,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
             # put the result on the result queue
             display.debug("sending task result for task %s" % self._task._uuid)
             try:
-                self._final_q.send_task_result(TaskResult(
+                self._final_q.send_task_result(_RawTaskResult(
                     host=self._host,
                     task=self._task,
                     return_data=executor_result,
@@ -267,7 +267,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
                 try:
                     raise AnsibleError("Task result omitted due to queue send failure.") from ex
                 except Exception as ex_wrapper:
-                    self._final_q.send_task_result(TaskResult(
+                    self._final_q.send_task_result(_RawTaskResult(
                         host=self._host,
                         task=self._task,
                         return_data=ActionBase.result_dict_from_exception(ex_wrapper),  # Overriding the task result, to represent the failure
@@ -283,7 +283,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
 
             self._host.vars = dict()
             self._host.groups = []
-            self._final_q.send_task_result(TaskResult(
+            self._final_q.send_task_result(_RawTaskResult(
                 host=self._host,
                 task=self._task,
                 return_data=return_data,
@@ -295,7 +295,7 @@ class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defin
                 try:
                     self._host.vars = dict()
                     self._host.groups = []
-                    self._final_q.send_task_result(TaskResult(
+                    self._final_q.send_task_result(_RawTaskResult(
                         host=self._host,
                         task=self._task,
                         return_data=ActionBase.result_dict_from_exception(ex),
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 21e492401d4..5e80c3d7296 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -20,7 +20,7 @@ from ansible.errors import (
     AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleConnectionFailure, AnsibleActionFail, AnsibleActionSkip, AnsibleTaskError,
     AnsibleValueOmittedError,
 )
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import _RawTaskResult
 from ansible._internal._datatag import _utils
 from ansible.module_utils._internal._plugin_exec_context import PluginExecContext
 from ansible.module_utils.common.messages import Detail, WarningSummary, DeprecationSummary
@@ -364,7 +364,7 @@ class TaskExecutor:
             if self._connection and not isinstance(self._connection, string_types):
                 task_fields['connection'] = getattr(self._connection, 'ansible_name')
 
-            tr = TaskResult(
+            tr = _RawTaskResult(
                 host=self._host,
                 task=self._task,
                 return_data=res,
@@ -669,7 +669,7 @@ class TaskExecutor:
                     if result.get('failed'):
                         self._final_q.send_callback(
                             'v2_runner_on_async_failed',
-                            TaskResult(
+                            _RawTaskResult(
                                 host=self._host,
                                 task=self._task,
                                 return_data=result,
@@ -679,7 +679,7 @@ class TaskExecutor:
                     else:
                         self._final_q.send_callback(
                             'v2_runner_on_async_ok',
-                            TaskResult(
+                            _RawTaskResult(
                                 host=self._host,
                                 task=self._task,
                                 return_data=result,
@@ -765,7 +765,7 @@ class TaskExecutor:
                         display.debug('Retrying task, attempt %d of %d' % (attempt, retries))
                         self._final_q.send_callback(
                             'v2_runner_retry',
-                            TaskResult(
+                            _RawTaskResult(
                                 host=self._host,
                                 task=self._task,
                                 return_data=result,
@@ -935,7 +935,7 @@ class TaskExecutor:
                 time_left -= self._task.poll
                 self._final_q.send_callback(
                     'v2_runner_on_async_poll',
-                    TaskResult(
+                    _RawTaskResult(
                         host=self._host,
                         task=async_task,
                         return_data=async_result,
diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py
index c26b9896cd6..9b1a83feb36 100644
--- a/lib/ansible/executor/task_queue_manager.py
+++ b/lib/ansible/executor/task_queue_manager.py
@@ -32,7 +32,7 @@ from ansible.errors import AnsibleError, ExitCode, AnsibleCallbackError
 from ansible._internal._errors._handler import ErrorHandler
 from ansible.executor.play_iterator import PlayIterator
 from ansible.executor.stats import AggregateStats
-from ansible.executor.task_result import TaskResult, ThinTaskResult
+from ansible.executor.task_result import _RawTaskResult, _WireTaskResult
 from ansible.inventory.data import InventoryData
 from ansible.module_utils.six import string_types
 from ansible.module_utils.common.text.converters import to_native
@@ -48,7 +48,8 @@ from ansible.utils.display import Display
 from ansible.utils.lock import lock_decorator
 from ansible.utils.multiprocessing import context as multiprocessing_context
 
-from dataclasses import dataclass
+if t.TYPE_CHECKING:
+    from ansible.executor.process.worker import WorkerProcess
 
 __all__ = ['TaskQueueManager']
 
@@ -58,11 +59,13 @@ STDERR_FILENO = 2
 
 display = Display()
 
+_T = t.TypeVar('_T')
+
 
 @dataclasses.dataclass(frozen=True, kw_only=True, slots=True)
 class CallbackSend:
     method_name: str
-    thin_task_result: ThinTaskResult
+    wire_task_result: _WireTaskResult
 
 
 class DisplaySend:
@@ -72,7 +75,7 @@ class DisplaySend:
         self.kwargs = kwargs
 
 
-@dataclass
+@dataclasses.dataclass
 class PromptSend:
     worker_id: int
     prompt: str
@@ -87,11 +90,11 @@ class FinalQueue(multiprocessing.queues.SimpleQueue):
         kwargs['ctx'] = multiprocessing_context
         super().__init__(*args, **kwargs)
 
-    def send_callback(self, method_name: str, task_result: TaskResult) -> None:
-        self.put(CallbackSend(method_name=method_name, thin_task_result=task_result.as_thin()))
+    def send_callback(self, method_name: str, task_result: _RawTaskResult) -> None:
+        self.put(CallbackSend(method_name=method_name, wire_task_result=task_result.as_wire_task_result()))
 
-    def send_task_result(self, task_result: TaskResult) -> None:
-        self.put(task_result.as_thin())
+    def send_task_result(self, task_result: _RawTaskResult) -> None:
+        self.put(task_result.as_wire_task_result())
 
     def send_display(self, method, *args, **kwargs):
         self.put(
@@ -186,11 +189,8 @@ class TaskQueueManager:
         # 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 _initialize_processes(self, num: int) -> None:
+        self._workers: list[WorkerProcess | None] = [None] * num
 
     def load_callbacks(self):
         """
@@ -430,54 +430,72 @@ class TaskQueueManager:
                 defunct = True
         return defunct
 
+    @staticmethod
+    def _first_arg_of_type(value_type: t.Type[_T], args: t.Sequence) -> _T | None:
+        return next((arg for arg in args if isinstance(arg, value_type)), None)
+
     @lock_decorator(attr='_callback_lock')
     def send_callback(self, method_name, *args, **kwargs):
         # We always send events to stdout callback first, rest should follow config order
         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):
+            if callback_plugin.disabled:
                 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)
+            if not callback_plugin.wants_implicit_tasks and (task_arg := self._first_arg_of_type(Task, args)) and task_arg.implicit:
+                continue
 
             # 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)
+                method = getattr(callback_plugin, possible, None)
 
-                if isinstance(arg, Task) and arg.implicit:
-                    is_implicit_task = True
+                if method is None:
+                    method = getattr(callback_plugin, possible.removeprefix('v2_'), None)
 
-            if is_implicit_task and not wants_implicit_tasks:
-                continue
+                    if method is not None:
+                        display.deprecated(
+                            msg='The v1 callback API is deprecated.',
+                            version='2.23',
+                            help_text='Use `v2_` prefixed callback methods instead.',
+                        )
+
+                if method is not None and not getattr(method, '_base_impl', False):  # don't bother dispatching to the base impls
+                    if possible == 'v2_on_any':
+                        display.deprecated(
+                            msg='The `v2_on_any` callback method is deprecated.',
+                            version='2.23',
+                            help_text='Use event-specific callback methods instead.',
+                        )
+
+                    methods.append(method)
 
             for method in methods:
+                # send clean copies
+                new_args = []
+
+                for arg in args:
+                    # FIXME: add play/task cleaners
+                    if isinstance(arg, _RawTaskResult):
+                        copied_tr = arg.as_callback_task_result()
+                        new_args.append(copied_tr)
+                        # this state hack requires that no callback ever accepts > 1 TaskResult object
+                        callback_plugin._current_task_result = copied_tr
+                    else:
+                        new_args.append(arg)
+
                 with self._callback_dispatch_error_handler.handle(AnsibleCallbackError):
                     try:
                         method(*new_args, **kwargs)
+                    except AssertionError:
+                        # Using an `assert` in integration tests is useful.
+                        # Production code should never use `assert` or raise `AssertionError`.
+                        raise
                     except Exception as ex:
                         raise AnsibleCallbackError(f"Callback dispatch {method_name!r} failed for plugin {callback_plugin._load_name!r}.") from ex
+
+            callback_plugin._current_task_result = None
diff --git a/lib/ansible/executor/task_result.py b/lib/ansible/executor/task_result.py
index c1c76e0da52..a60c97cab17 100644
--- a/lib/ansible/executor/task_result.py
+++ b/lib/ansible/executor/task_result.py
@@ -4,19 +4,24 @@
 
 from __future__ import annotations
 
+import collections.abc as _c
 import dataclasses
+import functools
 import typing as t
 
-from ansible import constants as C
+from ansible import constants
+from ansible.utils import vars as _vars
 from ansible.vars.clean import module_response_deepcopy, strip_internal_keys
+from ansible.module_utils.common import messages as _messages
+from ansible._internal import _collection_proxy
 
 if t.TYPE_CHECKING:
     from ansible.inventory.host import Host
     from ansible.playbook.task import Task
 
 _IGNORE = ('failed', 'skipped')
-_PRESERVE = ('attempts', 'changed', 'retries', '_ansible_no_log')
-_SUB_PRESERVE = {'_ansible_delegated_vars': ('ansible_host', 'ansible_port', 'ansible_user', 'ansible_connection')}
+_PRESERVE = {'attempts', 'changed', 'retries', '_ansible_no_log'}
+_SUB_PRESERVE = {'_ansible_delegated_vars': {'ansible_host', 'ansible_port', 'ansible_user', 'ansible_connection'}}
 
 # stuff callbacks need
 CLEAN_EXCEPTIONS = (
@@ -27,72 +32,120 @@ CLEAN_EXCEPTIONS = (
 )
 
 
+@t.final
 @dataclasses.dataclass(frozen=True, kw_only=True, slots=True)
-class ThinTaskResult:
-    """A thin version of `TaskResult` which can be sent over the worker queue."""
+class _WireTaskResult:
+    """A thin version of `_RawTaskResult` which can be sent over the worker queue."""
 
     host_name: str
     task_uuid: str
-    return_data: dict[str, object]
-    task_fields: dict[str, object]
+    return_data: _c.MutableMapping[str, object]
+    task_fields: _c.Mapping[str, object]
 
 
-class TaskResult:
+class _BaseTaskResult:
     """
     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: Host, task: Task, return_data: dict[str, object], task_fields: dict[str, object]) -> None:
-        self._host = host
-        self._task = task
-        self._result = return_data
-        self._task_fields = task_fields
-
-    def as_thin(self) -> ThinTaskResult:
-        """Return a `ThinTaskResult` from this instance."""
-        return ThinTaskResult(
-            host_name=self._host.name,
-            task_uuid=self._task._uuid,
-            return_data=self._result,
-            task_fields=self._task_fields,
-        )
+    def __init__(self, host: Host, task: Task, return_data: _c.MutableMapping[str, t.Any], task_fields: _c.Mapping[str, t.Any]) -> None:
+        self.__host = host
+        self.__task = task
+        self._return_data = return_data  # FIXME: this should be immutable, but strategy result processing mutates it in some corner cases
+        self.__task_fields = task_fields
+
+    @property
+    def host(self) -> Host:
+        """The host associated with this result."""
+        return self.__host
+
+    @property
+    def _host(self) -> Host:
+        """Use the `host` property when supporting only ansible-core 2.19 or later."""
+        # deprecated: description='Deprecate `_host` in favor of `host`' core_version='2.23'
+        return self.__host
+
+    @property
+    def task(self) -> Task:
+        """The task associated with this result."""
+        return self.__task
+
+    @property
+    def _task(self) -> Task:
+        """Use the `task` property when supporting only ansible-core 2.19 or later."""
+        # deprecated: description='Deprecate `_task` in favor of `task`' core_version='2.23'
+        return self.__task
+
+    @property
+    def task_fields(self) -> _c.Mapping[str, t.Any]:
+        """The task fields associated with this result."""
+        return self.__task_fields
+
+    @property
+    def _task_fields(self) -> _c.Mapping[str, t.Any]:
+        """Use the `task_fields` property when supporting only ansible-core 2.19 or later."""
+        # deprecated: description='Deprecate `_task_fields` in favor of `task`' core_version='2.23'
+        return self.__task_fields
+
+    @property
+    def exception(self) -> _messages.ErrorSummary | None:
+        """The error from this task result, if any."""
+        return self._return_data.get('exception')
+
+    @property
+    def warnings(self) -> _c.Sequence[_messages.WarningSummary]:
+        """The warnings for this task, if any."""
+        return _collection_proxy.SequenceProxy(self._return_data.get('warnings') or [])
+
+    @property
+    def deprecations(self) -> _c.Sequence[_messages.DeprecationSummary]:
+        """The deprecation warnings for this task, if any."""
+        return _collection_proxy.SequenceProxy(self._return_data.get('deprecations') or [])
+
+    @property
+    def _loop_results(self) -> list[_c.MutableMapping[str, t.Any]]:
+        """Return a list of loop results. If no loop results are present, an empty list is returned."""
+        results = self._return_data.get('results')
+
+        if not isinstance(results, list):
+            return []
+
+        return results
 
     @property
-    def task_name(self):
-        return self._task_fields.get('name', None) or self._task.get_name()
+    def task_name(self) -> str:
+        return str(self.task_fields.get('name', '')) or self.task.get_name()
 
-    def is_changed(self):
+    def is_changed(self) -> bool:
         return self._check_key('changed')
 
-    def is_skipped(self):
-        # loop results
-        if 'results' in self._result:
-            results = self._result['results']
+    def is_skipped(self) -> bool:
+        if self._loop_results:
             # Loop tasks are only considered skipped if all items were skipped.
             # some squashed results (eg, dnf) 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):
+            if all(isinstance(loop_res, dict) and loop_res.get('skipped', False) for loop_res in self._loop_results):
                 return True
 
         # regular tasks and squashed non-dict results
-        return self._result.get('skipped', False)
+        return bool(self._return_data.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]:
+    def is_failed(self) -> bool:
+        if 'failed_when_result' in self._return_data or any(isinstance(loop_res, dict) and 'failed_when_result' in loop_res for loop_res in self._loop_results):
             return self._check_key('failed_when_result')
-        else:
-            return self._check_key('failed')
 
-    def is_unreachable(self):
+        return self._check_key('failed')
+
+    def is_unreachable(self) -> bool:
         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')
+    def needs_debugger(self, globally_enabled: bool = False) -> bool:
+        _debugger = self.task_fields.get('debugger')
+        _ignore_errors = constants.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
 
@@ -109,68 +162,96 @@ class TaskResult:
 
         return ret
 
-    def _check_key(self, key):
-        """get a specific key from the result or its items"""
+    def _check_key(self, key: str) -> bool:
+        """Fetch a specific named boolean value from the result; if missing, a logical OR of the value from nested loop results; False for non-loop results."""
+        if (value := self._return_data.get(key, ...)) is not ...:
+            return bool(value)
 
-        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
+        return any(isinstance(result, dict) and result.get(key) for result in self._loop_results)
 
-    def clean_copy(self):
 
-        """ returns 'clean' taskresult object """
+@t.final
+class _RawTaskResult(_BaseTaskResult):
+    def as_wire_task_result(self) -> _WireTaskResult:
+        """Return a `_WireTaskResult` from this instance."""
+        return _WireTaskResult(
+            host_name=self.host.name,
+            task_uuid=self.task._uuid,
+            return_data=self._return_data,
+            task_fields=self.task_fields,
+        )
 
-        # FIXME: clean task_fields, _task and _host copies
-        result = TaskResult(self._host, self._task, {}, self._task_fields)
+    def as_callback_task_result(self) -> CallbackTaskResult:
+        """Return a `CallbackTaskResult` from this instance."""
+        ignore: tuple[str, ...]
 
         # statuses are already reflected on the event type
-        if result._task and result._task.action in C._ACTION_DEBUG:
+        if self.task and self.task.action in constants._ACTION_DEBUG:
             # debug is verbose by default to display vars, no need to add invocation
             ignore = _IGNORE + ('invocation',)
         else:
             ignore = _IGNORE
 
-        subset = {}
+        subset: dict[str, dict[str, object]] = {}
+
         # 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]
+        for sub, sub_keys in _SUB_PRESERVE.items():
+            sub_data = self._return_data.get(sub)
+
+            if isinstance(sub_data, dict):
+                subset[sub] = {key: value for key, value in sub_data.items() if key in sub_keys}
 
         # DTFIX-FUTURE: is checking no_log here redundant now that we use _ansible_no_log everywhere?
-        if isinstance(self._task.no_log, bool) and self._task.no_log or self._result.get('_ansible_no_log'):
-            censored_result = censor_result(self._result)
+        if isinstance(self.task.no_log, bool) and self.task.no_log or self._return_data.get('_ansible_no_log'):
+            censored_result = censor_result(self._return_data)
 
-            if results := self._result.get('results'):
+            if self._loop_results:
                 # maintain shape for loop results so callback behavior recognizes a loop was performed
-                censored_result.update(results=[censor_result(item) if item.get('_ansible_no_log') else item for item in results])
+                censored_result.update(results=[
+                    censor_result(loop_res) if isinstance(loop_res, dict) and loop_res.get('_ansible_no_log') else loop_res for loop_res in self._loop_results
+                ])
 
-            result._result = censored_result
-        elif self._result:
-            result._result = module_response_deepcopy(self._result)
-
-            # actually remove
-            for remove_key in ignore:
-                if remove_key in result._result:
-                    del result._result[remove_key]
+            return_data = censored_result
+        elif self._return_data:
+            return_data = {k: v for k, v in module_response_deepcopy(self._return_data).items() if k not in ignore}
 
             # remove almost ALL internal keys, keep ones relevant to callback
-            strip_internal_keys(result._result, exceptions=CLEAN_EXCEPTIONS)
+            strip_internal_keys(return_data, exceptions=CLEAN_EXCEPTIONS)
+        else:
+            return_data = {}
 
         # keep subset
-        result._result.update(subset)
+        return_data.update(subset)
+
+        return CallbackTaskResult(self.host, self.task, return_data, self.task_fields)
+
+
+@t.final
+class CallbackTaskResult(_BaseTaskResult):
+    """Public contract of TaskResult """
+
+    # DTFIX-RELEASE: find a better home for this since it's public API
+
+    @property
+    def _result(self) -> _c.MutableMapping[str, t.Any]:
+        """Use the `result` property when supporting only ansible-core 2.19 or later."""
+        # deprecated: description='Deprecate `_result` in favor of `result`' core_version='2.23'
+        return self.result
+
+    @functools.cached_property
+    def result(self) -> _c.MutableMapping[str, t.Any]:
+        """
+        Returns a cached copy of the task result dictionary for consumption by callbacks.
+        Internal custom types are transformed to native Python types to facilitate access and serialization.
+        """
+        return t.cast(_c.MutableMapping[str, t.Any], _vars.transform_to_native_types(self._return_data))
+
 
-        return result
+TaskResult = CallbackTaskResult
+"""Compatibility name for the pre-2.19 callback-shaped TaskResult passed to callbacks."""
 
 
-def censor_result(result: dict[str, t.Any]) -> dict[str, t.Any]:
+def censor_result(result: _c.Mapping[str, t.Any]) -> dict[str, t.Any]:
     censored_result = {key: value for key in _PRESERVE if (value := result.get(key, ...)) is not ...}
     censored_result.update(censored="the output has been hidden due to the fact that 'no_log: true' was specified for this result")
 
diff --git a/lib/ansible/inventory/manager.py b/lib/ansible/inventory/manager.py
index 914be9bd305..beff92a4b14 100644
--- a/lib/ansible/inventory/manager.py
+++ b/lib/ansible/inventory/manager.py
@@ -30,6 +30,7 @@ from random import shuffle
 
 from ansible import constants as C
 from ansible._internal import _json, _wrapt
+from ansible._internal._json import EncryptedStringBehavior
 from ansible.errors import AnsibleError, AnsibleOptionsError
 from ansible.inventory.data import InventoryData
 from ansible.module_utils.six import string_types
@@ -787,7 +788,7 @@ class _InventoryDataWrapper(_wrapt.ObjectProxy):
         return _json.AnsibleVariableVisitor(
             trusted_as_template=self._target_plugin.trusted_by_default,
             origin=self._default_origin,
-            allow_encrypted_string=True,
+            encrypted_string_behavior=EncryptedStringBehavior.PRESERVE,
         )
 
     def set_variable(self, entity: str, varname: str, value: t.Any) -> None:
diff --git a/lib/ansible/playbook/included_file.py b/lib/ansible/playbook/included_file.py
index 673f5cfd71f..a8386509b76 100644
--- a/lib/ansible/playbook/included_file.py
+++ b/lib/ansible/playbook/included_file.py
@@ -21,34 +21,42 @@ import os
 
 from ansible import constants as C
 from ansible.errors import AnsibleError
+from ansible.executor.task_result import _RawTaskResult
+from ansible.inventory.host import Host
 from ansible.module_utils.common.text.converters import to_text
+from ansible.parsing.dataloader import DataLoader
 from ansible.playbook.handler import Handler
 from ansible.playbook.task_include import TaskInclude
 from ansible.playbook.role_include import IncludeRole
 from ansible._internal._templating._engine import TemplateEngine
 from ansible.utils.display import Display
+from ansible.vars.manager import VariableManager
 
 display = Display()
 
 
 class IncludedFile:
 
-    def __init__(self, filename, args, vars, task, is_role=False):
+    def __init__(self, filename, args, vars, task, is_role: bool = False) -> None:
         self._filename = filename
         self._args = args
         self._vars = vars
         self._task = task
-        self._hosts = []
+        self._hosts: list[Host] = []
         self._is_role = is_role
-        self._results = []
+        self._results: list[_RawTaskResult] = []
 
-    def add_host(self, host):
+    def add_host(self, host: Host) -> None:
         if host not in self._hosts:
             self._hosts.append(host)
             return
+
         raise ValueError()
 
     def __eq__(self, other):
+        if not isinstance(other, IncludedFile):
+            return False
+
         return (other._filename == self._filename and
                 other._args == self._args and
                 other._vars == self._vars and
@@ -59,23 +67,28 @@ class IncludedFile:
         return "%s (args=%s vars=%s): %s" % (self._filename, self._args, self._vars, self._hosts)
 
     @staticmethod
-    def process_include_results(results, iterator, loader, variable_manager):
-        included_files = []
-        task_vars_cache = {}
+    def process_include_results(
+        results: list[_RawTaskResult],
+        iterator,
+        loader: DataLoader,
+        variable_manager: VariableManager,
+    ) -> list[IncludedFile]:
+        included_files: list[IncludedFile] = []
+        task_vars_cache: dict[tuple, dict] = {}
 
         for res in results:
 
-            original_host = res._host
-            original_task = res._task
+            original_host = res.host
+            original_task = res.task
 
             if original_task.action in C._ACTION_ALL_INCLUDES:
 
                 if original_task.loop:
-                    if 'results' not in res._result:
+                    if 'results' not in res._return_data:
                         continue
-                    include_results = res._result['results']
+                    include_results = res._loop_results
                 else:
-                    include_results = [res._result]
+                    include_results = [res._return_data]
 
                 for include_result in include_results:
                     # if the task result was skipped or failed, continue
diff --git a/lib/ansible/plugins/callback/__init__.py b/lib/ansible/plugins/callback/__init__.py
index f88055a4daa..1d2ed693cc3 100644
--- a/lib/ansible/plugins/callback/__init__.py
+++ b/lib/ansible/plugins/callback/__init__.py
@@ -24,15 +24,14 @@ import re
 import sys
 import textwrap
 import typing as t
+import collections.abc as _c
 
 from typing import TYPE_CHECKING
 
-from collections.abc import MutableMapping
 from copy import deepcopy
 
 from ansible import constants as C
 from ansible.module_utils._internal import _datatag
-from ansible.module_utils.common.messages import ErrorSummary
 from ansible._internal._yaml import _dumper
 from ansible.plugins import AnsiblePlugin
 from ansible.utils.color import stringc
@@ -44,7 +43,7 @@ from ansible._internal._templating import _engine
 import yaml
 
 if TYPE_CHECKING:
-    from ansible.executor.task_result import TaskResult
+    from ansible.executor.task_result import CallbackTaskResult
 
 global_display = Display()
 
@@ -59,6 +58,19 @@ _YAML_BREAK_CHARS = '\n\x85\u2028\u2029'  # NL, NEL, LS, PS
 _SPACE_BREAK_RE = re.compile(fr' +([{_YAML_BREAK_CHARS}])')
 
 
+_T_callable = t.TypeVar("_T_callable", bound=t.Callable)
+
+
+def _callback_base_impl(wrapped: _T_callable) -> _T_callable:
+    """
+    Decorator for the no-op methods on the `CallbackBase` base class.
+    Used to avoid unnecessary dispatch overhead to no-op base callback methods.
+    """
+    wrapped._base_impl = True
+
+    return wrapped
+
+
 class _AnsibleCallbackDumper(_dumper.AnsibleDumper):
     def __init__(self, *args, lossy: bool = False, **kwargs):
         super().__init__(*args, **kwargs)
@@ -87,6 +99,8 @@ class _AnsibleCallbackDumper(_dumper.AnsibleDumper):
     def _register_representers(cls) -> None:
         super()._register_representers()
 
+        # exact type checks occur first against representers, then subclasses against multi-representers
+        cls.add_representer(str, cls._pretty_represent_str)
         cls.add_multi_representer(str, cls._pretty_represent_str)
 
 
@@ -140,12 +154,17 @@ class CallbackBase(AnsiblePlugin):
     custom actions.
     """
 
-    def __init__(self, display=None, options=None):
+    def __init__(self, display: Display | None = None, options: dict[str, t.Any] | None = None) -> None:
+        super().__init__()
+
         if display:
             self._display = display
         else:
             self._display = global_display
 
+        # FUTURE: fix double-loading of non-collection stdout callback plugins that don't set CALLBACK_NEEDS_ENABLED
+
+        # FUTURE: this code is jacked for 2.x- it should just use the type names and always assume 2.0+ for normal cases
         if self._display.verbosity >= 4:
             name = getattr(self, 'CALLBACK_NAME', 'unnamed')
             ctype = getattr(self, 'CALLBACK_TYPE', 'old')
@@ -155,7 +174,8 @@ class CallbackBase(AnsiblePlugin):
         self.disabled = False
         self.wants_implicit_tasks = False
 
-        self._plugin_options = {}
+        self._plugin_options: dict[str, t.Any] = {}
+
         if options is not None:
             self.set_options(options)
 
@@ -164,6 +184,8 @@ class CallbackBase(AnsiblePlugin):
             'ansible_loop_var', 'ansible_index_var', 'ansible_loop',
         )
 
+        self._current_task_result: CallbackTaskResult | None = None
+
     # helper for callbacks, so they don't all have to include deepcopy
     _copy_result = deepcopy
 
@@ -185,25 +207,30 @@ class CallbackBase(AnsiblePlugin):
         self._plugin_options = C.config.get_plugin_options(self.plugin_type, self._load_name, keys=task_keys, variables=var_options, direct=direct)
 
     @staticmethod
-    def host_label(result):
-        """Return label for the hostname (& delegated hostname) of a task
-        result.
-        """
-        label = "%s" % result._host.get_name()
-        if result._task.delegate_to and result._task.delegate_to != result._host.get_name():
+    def host_label(result: CallbackTaskResult) -> str:
+        """Return label for the hostname (& delegated hostname) of a task result."""
+        label = result.host.get_name()
+        if result.task.delegate_to and result.task.delegate_to != result.host.get_name():
             # show delegated host
-            label += " -> %s" % result._task.delegate_to
+            label += " -> %s" % result.task.delegate_to
             # in case we have 'extra resolution'
-            ahost = result._result.get('_ansible_delegated_vars', {}).get('ansible_host', result._task.delegate_to)
-            if result._task.delegate_to != ahost:
+            ahost = result.result.get('_ansible_delegated_vars', {}).get('ansible_host', result.task.delegate_to)
+            if result.task.delegate_to != ahost:
                 label += "(%s)" % ahost
         return label
 
-    def _run_is_verbose(self, result, verbosity=0):
-        return ((self._display.verbosity > verbosity or result._result.get('_ansible_verbose_always', False) is True)
-                and result._result.get('_ansible_verbose_override', False) is False)
-
-    def _dump_results(self, result, indent=None, sort_keys=True, keep_invocation=False, serialize=True):
+    def _run_is_verbose(self, result: CallbackTaskResult, verbosity: int = 0) -> bool:
+        return ((self._display.verbosity > verbosity or result.result.get('_ansible_verbose_always', False) is True)
+                and result.result.get('_ansible_verbose_override', False) is False)
+
+    def _dump_results(
+        self,
+        result: _c.Mapping[str, t.Any],
+        indent: int | None = None,
+        sort_keys: bool = True,
+        keep_invocation: bool = False,
+        serialize: bool = True,
+    ) -> str:
         try:
             result_format = self.get_option('result_format')
         except KeyError:
@@ -253,10 +280,12 @@ class CallbackBase(AnsiblePlugin):
             # that want to further modify the result, or use custom serialization
             return abridged_result
 
+        # DTFIX-RELEASE: Switch to stock json/yaml serializers here? We should always have a transformed plain-types result.
+
         if result_format == 'json':
             return json.dumps(abridged_result, cls=_fallback_to_str.Encoder, indent=indent, ensure_ascii=False, sort_keys=sort_keys)
 
-        elif result_format == 'yaml':
+        if result_format == 'yaml':
             # None is a sentinel in this case that indicates default behavior
             # default behavior for yaml is to prettify results
             lossy = pretty_results in (None, True)
@@ -281,22 +310,28 @@ class CallbackBase(AnsiblePlugin):
                 ' ' * (indent or 4)
             )
 
-    def _handle_warnings(self, res: dict[str, t.Any]) -> None:
-        """Display warnings and deprecation warnings sourced by task execution."""
-        for warning in res.pop('warnings', []):
-            # DTFIX-RELEASE: what to do about propagating wrap_text from the original display.warning call?
-            self._display._warning(warning, wrap_text=False)
-
-        for warning in res.pop('deprecations', []):
-            self._display._deprecated(warning)
+        # DTFIX-RELEASE: add test to exercise this case
+        raise ValueError(f'Unsupported result_format {result_format!r}.')
 
-    def _handle_exception(self, result: dict[str, t.Any], use_stderr: bool = False) -> None:
-        error_summary: ErrorSummary | None
-
-        if error_summary := result.pop('exception', None):
-            self._display._error(error_summary, wrap_text=False, stderr=use_stderr)
-
-    def _handle_warnings_and_exception(self, result: TaskResult) -> None:
+    def _handle_warnings(self, res: _c.MutableMapping[str, t.Any]) -> None:
+        """Display warnings and deprecation warnings sourced by task execution."""
+        if res.pop('warnings', None) and self._current_task_result and (warnings := self._current_task_result.warnings):
+            # display warnings from the current task result if `warnings` was not removed from `result` (or made falsey)
+            for warning in warnings:
+                # DTFIX-RELEASE: what to do about propagating wrap_text from the original display.warning call?
+                self._display._warning(warning, wrap_text=False)
+
+        if res.pop('deprecations', None) and self._current_task_result and (deprecations := self._current_task_result.deprecations):
+            # display deprecations from the current task result if `deprecations` was not removed from `result` (or made falsey)
+            for deprecation in deprecations:
+                self._display._deprecated(deprecation)
+
+    def _handle_exception(self, result: _c.MutableMapping[str, t.Any], use_stderr: bool = False) -> None:
+        if result.pop('exception', None) and self._current_task_result and (exception := self._current_task_result.exception):
+            # display exception from the current task result if `exception` was not removed from `result` (or made falsey)
+            self._display._error(exception, wrap_text=False, stderr=use_stderr)
+
+    def _handle_warnings_and_exception(self, result: CallbackTaskResult) -> None:
         """Standardized handling of warnings/deprecations and exceptions from a task/item result."""
         # DTFIX-RELEASE: make/doc/porting-guide a public version of this method?
         try:
@@ -304,8 +339,8 @@ class CallbackBase(AnsiblePlugin):
         except KeyError:
             use_stderr = False
 
-        self._handle_warnings(result._result)
-        self._handle_exception(result._result, use_stderr=use_stderr)
+        self._handle_warnings(result.result)
+        self._handle_exception(result.result, use_stderr=use_stderr)
 
     def _serialize_diff(self, diff):
         try:
@@ -322,7 +357,8 @@ class CallbackBase(AnsiblePlugin):
 
         if result_format == 'json':
             return json.dumps(diff, sort_keys=True, indent=4, separators=(u',', u': ')) + u'\n'
-        elif result_format == 'yaml':
+
+        if result_format == 'yaml':
             # None is a sentinel in this case that indicates default behavior
             # default behavior for yaml is to prettify results
             lossy = pretty_results in (None, True)
@@ -338,6 +374,9 @@ class CallbackBase(AnsiblePlugin):
                 '    '
             )
 
+        # DTFIX-RELEASE: add test to exercise this case
+        raise ValueError(f'Unsupported result_format {result_format!r}.')
+
     def _get_diff(self, difflist):
 
         if not isinstance(difflist, list):
@@ -356,7 +395,7 @@ class CallbackBase(AnsiblePlugin):
             if 'before' in diff and 'after' in diff:
                 # format complex structures into 'files'
                 for x in ['before', 'after']:
-                    if isinstance(diff[x], MutableMapping):
+                    if isinstance(diff[x], _c.Mapping):
                         diff[x] = self._serialize_diff(diff[x])
                     elif diff[x] is None:
                         diff[x] = ''
@@ -398,7 +437,7 @@ class CallbackBase(AnsiblePlugin):
                 ret.append(diff['prepared'])
         return u''.join(ret)
 
-    def _get_item_label(self, result):
+    def _get_item_label(self, result: _c.Mapping[str, t.Any]) -> t.Any:
         """ retrieves the value to be displayed as a label for an item entry from a result object"""
         if result.get('_ansible_no_log', False):
             item = "(censored due to no_log)"
@@ -406,9 +445,9 @@ class CallbackBase(AnsiblePlugin):
             item = result.get('_ansible_item_label', result.get('item'))
         return item
 
-    def _process_items(self, result):
+    def _process_items(self, result: CallbackTaskResult) -> None:
         # just remove them as now they get handled by individual callbacks
-        del result._result['results']
+        del result.result['results']
 
     def _clean_results(self, result, task_name):
         """ removes data from results for display """
@@ -434,74 +473,97 @@ class CallbackBase(AnsiblePlugin):
     def set_play_context(self, play_context):
         pass
 
+    @_callback_base_impl
     def on_any(self, *args, **kwargs):
         pass
 
+    @_callback_base_impl
     def runner_on_failed(self, host, res, ignore_errors=False):
         pass
 
+    @_callback_base_impl
     def runner_on_ok(self, host, res):
         pass
 
+    @_callback_base_impl
     def runner_on_skipped(self, host, item=None):
         pass
 
+    @_callback_base_impl
     def runner_on_unreachable(self, host, res):
         pass
 
+    @_callback_base_impl
     def runner_on_no_hosts(self):
         pass
 
+    @_callback_base_impl
     def runner_on_async_poll(self, host, res, jid, clock):
         pass
 
+    @_callback_base_impl
     def runner_on_async_ok(self, host, res, jid):
         pass
 
+    @_callback_base_impl
     def runner_on_async_failed(self, host, res, jid):
         pass
 
+    @_callback_base_impl
     def playbook_on_start(self):
         pass
 
+    @_callback_base_impl
     def playbook_on_notify(self, host, handler):
         pass
 
+    @_callback_base_impl
     def playbook_on_no_hosts_matched(self):
         pass
 
+    @_callback_base_impl
     def playbook_on_no_hosts_remaining(self):
         pass
 
+    @_callback_base_impl
     def playbook_on_task_start(self, name, is_conditional):
         pass
 
+    @_callback_base_impl
     def playbook_on_vars_prompt(self, varname, private=True, prompt=None, encrypt=None, confirm=False, salt_size=None, salt=None, default=None, unsafe=None):
         pass
 
+    @_callback_base_impl
     def playbook_on_setup(self):
         pass
 
+    @_callback_base_impl
     def playbook_on_import_for_host(self, host, imported_file):
         pass
 
+    @_callback_base_impl
     def playbook_on_not_import_for_host(self, host, missing_file):
         pass
 
+    @_callback_base_impl
     def playbook_on_play_start(self, name):
         pass
 
+    @_callback_base_impl
     def playbook_on_stats(self, stats):
         pass
 
+    @_callback_base_impl
     def on_file_diff(self, host, diff):
         pass
 
     # V2 METHODS, by default they call v1 counterparts if possible
+    @_callback_base_impl
     def v2_on_any(self, *args, **kwargs):
         self.on_any(args, kwargs)
 
-    def v2_runner_on_failed(self, result: TaskResult, ignore_errors: bool = False) -> None:
+    @_callback_base_impl
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
         """Process results of a failed task.
 
         Note: The value of 'ignore_errors' tells Ansible whether to
@@ -512,7 +574,7 @@ class CallbackBase(AnsiblePlugin):
         issues (for example, missing packages), or syntax errors.
 
         :param result: The parameters of the task and its results.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
         :param ignore_errors: Whether Ansible should continue \
             running tasks on the host where the task failed.
         :type ignore_errors: bool
@@ -520,147 +582,172 @@ class CallbackBase(AnsiblePlugin):
         :return: None
         :rtype: None
         """
-        host = result._host.get_name()
-        self.runner_on_failed(host, result._result, ignore_errors)
+        host = result.host.get_name()
+        self.runner_on_failed(host, result.result, ignore_errors)
 
-    def v2_runner_on_ok(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         """Process results of a successful task.
 
         :param result: The parameters of the task and its results.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
 
         :return: None
         :rtype: None
         """
-        host = result._host.get_name()
-        self.runner_on_ok(host, result._result)
+        host = result.host.get_name()
+        self.runner_on_ok(host, result.result)
 
-    def v2_runner_on_skipped(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
         """Process results of a skipped task.
 
         :param result: The parameters of the task and its results.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
 
         :return: None
         :rtype: None
         """
         if C.DISPLAY_SKIPPED_HOSTS:
-            host = result._host.get_name()
-            self.runner_on_skipped(host, self._get_item_label(getattr(result._result, 'results', {})))
+            host = result.host.get_name()
+            self.runner_on_skipped(host, self._get_item_label(getattr(result.result, 'results', {})))
 
-    def v2_runner_on_unreachable(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
         """Process results of a task if a target node is unreachable.
 
         :param result: The parameters of the task and its results.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
 
         :return: None
         :rtype: None
         """
-        host = result._host.get_name()
-        self.runner_on_unreachable(host, result._result)
+        host = result.host.get_name()
+        self.runner_on_unreachable(host, result.result)
 
-    def v2_runner_on_async_poll(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_on_async_poll(self, result: CallbackTaskResult) -> None:
         """Get details about an unfinished task running in async mode.
 
         Note: The value of the `poll` keyword in the task determines
         the interval at which polling occurs and this method is run.
 
         :param result: The parameters of the task and its status.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
 
         :rtype: None
         :rtype: None
         """
-        host = result._host.get_name()
-        jid = result._result.get('ansible_job_id')
+        host = result.host.get_name()
+        jid = result.result.get('ansible_job_id')
         # FIXME, get real clock
         clock = 0
-        self.runner_on_async_poll(host, result._result, jid, clock)
+        self.runner_on_async_poll(host, result.result, jid, clock)
 
-    def v2_runner_on_async_ok(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_on_async_ok(self, result: CallbackTaskResult) -> None:
         """Process results of a successful task that ran in async mode.
 
         :param result: The parameters of the task and its results.
-        :type result: TaskResult
+        :type result: CallbackTaskResult
 
         :return: None
         :rtype: None
         """
-        host = result._host.get_name()
-        jid = result._result.get('ansible_job_id')
-        self.runner_on_async_ok(host, result._result, jid)
+        host = result.host.get_name()
+        jid = result.result.get('ansible_job_id')
+        self.runner_on_async_ok(host, result.result, jid)
 
-    def v2_runner_on_async_failed(self, result):
-        host = result._host.get_name()
+    @_callback_base_impl
+    def v2_runner_on_async_failed(self, result: CallbackTaskResult) -> None:
+        host = result.host.get_name()
         # Attempt to get the async job ID. If the job does not finish before the
         # async timeout value, the ID may be within the unparsed 'async_result' dict.
-        jid = result._result.get('ansible_job_id')
-        if not jid and 'async_result' in result._result:
-            jid = result._result['async_result'].get('ansible_job_id')
-        self.runner_on_async_failed(host, result._result, jid)
+        jid = result.result.get('ansible_job_id')
+        if not jid and 'async_result' in result.result:
+            jid = result.result['async_result'].get('ansible_job_id')
+        self.runner_on_async_failed(host, result.result, jid)
 
+    @_callback_base_impl
     def v2_playbook_on_start(self, playbook):
         self.playbook_on_start()
 
+    @_callback_base_impl
     def v2_playbook_on_notify(self, handler, host):
         self.playbook_on_notify(host, handler)
 
+    @_callback_base_impl
     def v2_playbook_on_no_hosts_matched(self):
         self.playbook_on_no_hosts_matched()
 
+    @_callback_base_impl
     def v2_playbook_on_no_hosts_remaining(self):
         self.playbook_on_no_hosts_remaining()
 
+    @_callback_base_impl
     def v2_playbook_on_task_start(self, task, is_conditional):
         self.playbook_on_task_start(task.name, is_conditional)
 
     # FIXME: not called
+    @_callback_base_impl
     def v2_playbook_on_cleanup_task_start(self, task):
         pass  # no v1 correspondence
 
+    @_callback_base_impl
     def v2_playbook_on_handler_task_start(self, task):
         pass  # no v1 correspondence
 
+    @_callback_base_impl
     def v2_playbook_on_vars_prompt(self, varname, private=True, prompt=None, encrypt=None, confirm=False, salt_size=None, salt=None, default=None, unsafe=None):
         self.playbook_on_vars_prompt(varname, private, prompt, encrypt, confirm, salt_size, salt, default, unsafe)
 
     # FIXME: not called
-    def v2_playbook_on_import_for_host(self, result, imported_file):
-        host = result._host.get_name()
+    @_callback_base_impl
+    def v2_playbook_on_import_for_host(self, result: CallbackTaskResult, imported_file) -> None:
+        host = result.host.get_name()
         self.playbook_on_import_for_host(host, imported_file)
 
     # FIXME: not called
-    def v2_playbook_on_not_import_for_host(self, result, missing_file):
-        host = result._host.get_name()
+    @_callback_base_impl
+    def v2_playbook_on_not_import_for_host(self, result: CallbackTaskResult, missing_file) -> None:
+        host = result.host.get_name()
         self.playbook_on_not_import_for_host(host, missing_file)
 
+    @_callback_base_impl
     def v2_playbook_on_play_start(self, play):
         self.playbook_on_play_start(play.name)
 
+    @_callback_base_impl
     def v2_playbook_on_stats(self, stats):
         self.playbook_on_stats(stats)
 
-    def v2_on_file_diff(self, result):
-        if 'diff' in result._result:
-            host = result._host.get_name()
-            self.on_file_diff(host, result._result['diff'])
+    @_callback_base_impl
+    def v2_on_file_diff(self, result: CallbackTaskResult) -> None:
+        if 'diff' in result.result:
+            host = result.host.get_name()
+            self.on_file_diff(host, result.result['diff'])
 
+    @_callback_base_impl
     def v2_playbook_on_include(self, included_file):
         pass  # no v1 correspondence
 
-    def v2_runner_item_on_ok(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_item_on_ok(self, result: CallbackTaskResult) -> None:
         pass
 
-    def v2_runner_item_on_failed(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_item_on_failed(self, result: CallbackTaskResult) -> None:
         pass
 
-    def v2_runner_item_on_skipped(self, result: TaskResult) -> None:
+    @_callback_base_impl
+    def v2_runner_item_on_skipped(self, result: CallbackTaskResult) -> None:
         pass
 
-    def v2_runner_retry(self, result):
+    @_callback_base_impl
+    def v2_runner_retry(self, result: CallbackTaskResult) -> None:
         pass
 
+    @_callback_base_impl
     def v2_runner_on_start(self, host, task):
         """Event used when host begins execution of a task
 
diff --git a/lib/ansible/plugins/callback/default.py b/lib/ansible/plugins/callback/default.py
index 2237c73a759..5e029a16bed 100644
--- a/lib/ansible/plugins/callback/default.py
+++ b/lib/ansible/plugins/callback/default.py
@@ -21,7 +21,7 @@ DOCUMENTATION = """
 
 from ansible import constants as C
 from ansible import context
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import CallbackTaskResult
 from ansible.playbook.task_include import TaskInclude
 from ansible.plugins.callback import CallbackBase
 from ansible.utils.color import colorize, hostcolor
@@ -47,39 +47,39 @@ class CallbackModule(CallbackBase):
         self._task_type_cache = {}
         super(CallbackModule, self).__init__()
 
-    def v2_runner_on_failed(self, result: TaskResult, ignore_errors: bool = False) -> None:
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
         host_label = self.host_label(result)
 
-        if self._last_task_banner != result._task._uuid:
-            self._print_task_banner(result._task)
+        if self._last_task_banner != result.task._uuid:
+            self._print_task_banner(result.task)
 
         self._handle_warnings_and_exception(result)
 
         # FIXME: this method should not exist, delegate "suggested keys to display" to the plugin or something... As-is, the placement of this
         #          call obliterates `results`, which causes a task summary to be printed on loop failures, which we don't do anywhere else.
-        self._clean_results(result._result, result._task.action)
+        self._clean_results(result.result, result.task.action)
 
-        if result._task.loop and 'results' in result._result:
+        if result.task.loop and 'results' in result.result:
             self._process_items(result)
         else:
             if self._display.verbosity < 2 and self.get_option('show_task_path_on_failure'):
-                self._print_task_path(result._task)
-            msg = "fatal: [%s]: FAILED! => %s" % (host_label, self._dump_results(result._result))
+                self._print_task_path(result.task)
+            msg = "fatal: [%s]: FAILED! => %s" % (host_label, self._dump_results(result.result))
             self._display.display(msg, color=C.COLOR_ERROR, stderr=self.get_option('display_failed_stderr'))
 
         if ignore_errors:
             self._display.display("...ignoring", color=C.COLOR_SKIP)
 
-    def v2_runner_on_ok(self, result: TaskResult) -> None:
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         host_label = self.host_label(result)
 
-        if isinstance(result._task, TaskInclude):
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+        if isinstance(result.task, TaskInclude):
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
             return
-        elif result._result.get('changed', False):
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+        elif result.result.get('changed', False):
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             msg = "changed: [%s]" % (host_label,)
             color = C.COLOR_CHANGED
@@ -87,52 +87,52 @@ class CallbackModule(CallbackBase):
             if not self.get_option('display_ok_hosts'):
                 return
 
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             msg = "ok: [%s]" % (host_label,)
             color = C.COLOR_OK
 
         self._handle_warnings_and_exception(result)
 
-        if result._task.loop and 'results' in result._result:
+        if result.task.loop and 'results' in result.result:
             self._process_items(result)
         else:
-            self._clean_results(result._result, result._task.action)
+            self._clean_results(result.result, result.task.action)
 
             if self._run_is_verbose(result):
-                msg += " => %s" % (self._dump_results(result._result),)
+                msg += " => %s" % (self._dump_results(result.result),)
             self._display.display(msg, color=color)
 
-    def v2_runner_on_skipped(self, result: TaskResult) -> None:
+    def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
         if self.get_option('display_skipped_hosts'):
 
-            self._clean_results(result._result, result._task.action)
+            self._clean_results(result.result, result.task.action)
 
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             self._handle_warnings_and_exception(result)
 
-            if result._task.loop is not None and 'results' in result._result:
+            if result.task.loop is not None and 'results' in result.result:
                 self._process_items(result)
 
-            msg = "skipping: [%s]" % result._host.get_name()
+            msg = "skipping: [%s]" % result.host.get_name()
             if self._run_is_verbose(result):
-                msg += " => %s" % self._dump_results(result._result)
+                msg += " => %s" % self._dump_results(result.result)
             self._display.display(msg, color=C.COLOR_SKIP)
 
-    def v2_runner_on_unreachable(self, result: TaskResult) -> None:
-        if self._last_task_banner != result._task._uuid:
-            self._print_task_banner(result._task)
+    def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
+        if self._last_task_banner != result.task._uuid:
+            self._print_task_banner(result.task)
 
         self._handle_warnings_and_exception(result)
 
         host_label = self.host_label(result)
-        msg = "fatal: [%s]: UNREACHABLE! => %s" % (host_label, self._dump_results(result._result))
+        msg = "fatal: [%s]: UNREACHABLE! => %s" % (host_label, self._dump_results(result.result))
         self._display.display(msg, color=C.COLOR_UNREACHABLE, stderr=self.get_option('display_failed_stderr'))
 
-        if result._task.ignore_unreachable:
+        if result.task.ignore_unreachable:
             self._display.display("...ignoring", color=C.COLOR_SKIP)
 
     def v2_playbook_on_no_hosts_matched(self):
@@ -222,29 +222,29 @@ class CallbackModule(CallbackBase):
 
         self._display.banner(msg)
 
-    def v2_on_file_diff(self, result):
-        if result._task.loop and 'results' in result._result:
-            for res in result._result['results']:
+    def v2_on_file_diff(self, result: CallbackTaskResult) -> None:
+        if result.task.loop and 'results' in result.result:
+            for res in result.result['results']:
                 if 'diff' in res and res['diff'] and res.get('changed', False):
                     diff = self._get_diff(res['diff'])
                     if diff:
-                        if self._last_task_banner != result._task._uuid:
-                            self._print_task_banner(result._task)
+                        if self._last_task_banner != result.task._uuid:
+                            self._print_task_banner(result.task)
                         self._display.display(diff)
-        elif 'diff' in result._result and result._result['diff'] and result._result.get('changed', False):
-            diff = self._get_diff(result._result['diff'])
+        elif 'diff' in result.result and result.result['diff'] and result.result.get('changed', False):
+            diff = self._get_diff(result.result['diff'])
             if diff:
-                if self._last_task_banner != result._task._uuid:
-                    self._print_task_banner(result._task)
+                if self._last_task_banner != result.task._uuid:
+                    self._print_task_banner(result.task)
                 self._display.display(diff)
 
-    def v2_runner_item_on_ok(self, result: TaskResult) -> None:
+    def v2_runner_item_on_ok(self, result: CallbackTaskResult) -> None:
         host_label = self.host_label(result)
-        if isinstance(result._task, TaskInclude):
+        if isinstance(result.task, TaskInclude):
             return
-        elif result._result.get('changed', False):
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+        elif result.result.get('changed', False):
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             msg = 'changed'
             color = C.COLOR_CHANGED
@@ -252,47 +252,47 @@ class CallbackModule(CallbackBase):
             if not self.get_option('display_ok_hosts'):
                 return
 
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             msg = 'ok'
             color = C.COLOR_OK
 
         self._handle_warnings_and_exception(result)
 
-        msg = "%s: [%s] => (item=%s)" % (msg, host_label, self._get_item_label(result._result))
-        self._clean_results(result._result, result._task.action)
+        msg = "%s: [%s] => (item=%s)" % (msg, host_label, self._get_item_label(result.result))
+        self._clean_results(result.result, result.task.action)
         if self._run_is_verbose(result):
-            msg += " => %s" % self._dump_results(result._result)
+            msg += " => %s" % self._dump_results(result.result)
         self._display.display(msg, color=color)
 
-    def v2_runner_item_on_failed(self, result: TaskResult) -> None:
-        if self._last_task_banner != result._task._uuid:
-            self._print_task_banner(result._task)
+    def v2_runner_item_on_failed(self, result: CallbackTaskResult) -> None:
+        if self._last_task_banner != result.task._uuid:
+            self._print_task_banner(result.task)
 
         self._handle_warnings_and_exception(result)
 
         host_label = self.host_label(result)
 
         msg = "failed: [%s]" % (host_label,)
-        self._clean_results(result._result, result._task.action)
+        self._clean_results(result.result, result.task.action)
         self._display.display(
-            msg + " (item=%s) => %s" % (self._get_item_label(result._result), self._dump_results(result._result)),
+            msg + " (item=%s) => %s" % (self._get_item_label(result.result), self._dump_results(result.result)),
             color=C.COLOR_ERROR,
             stderr=self.get_option('display_failed_stderr')
         )
 
-    def v2_runner_item_on_skipped(self, result: TaskResult) -> None:
+    def v2_runner_item_on_skipped(self, result: CallbackTaskResult) -> None:
         if self.get_option('display_skipped_hosts'):
-            if self._last_task_banner != result._task._uuid:
-                self._print_task_banner(result._task)
+            if self._last_task_banner != result.task._uuid:
+                self._print_task_banner(result.task)
 
             self._handle_warnings_and_exception(result)
 
-            self._clean_results(result._result, result._task.action)
-            msg = "skipping: [%s] => (item=%s) " % (result._host.get_name(), self._get_item_label(result._result))
+            self._clean_results(result.result, result.task.action)
+            msg = "skipping: [%s] => (item=%s) " % (result.host.get_name(), self._get_item_label(result.result))
             if self._run_is_verbose(result):
-                msg += " => %s" % self._dump_results(result._result)
+                msg += " => %s" % self._dump_results(result.result)
             self._display.display(msg, color=C.COLOR_SKIP)
 
     def v2_playbook_on_include(self, included_file):
@@ -377,37 +377,37 @@ class CallbackModule(CallbackBase):
         if context.CLIARGS['check'] and self.get_option('check_mode_markers'):
             self._display.banner("DRY RUN")
 
-    def v2_runner_retry(self, result):
-        task_name = result.task_name or result._task
+    def v2_runner_retry(self, result: CallbackTaskResult) -> None:
+        task_name = result.task_name or result.task
         host_label = self.host_label(result)
-        msg = "FAILED - RETRYING: [%s]: %s (%d retries left)." % (host_label, task_name, result._result['retries'] - result._result['attempts'])
+        msg = "FAILED - RETRYING: [%s]: %s (%d retries left)." % (host_label, task_name, result.result['retries'] - result.result['attempts'])
         if self._run_is_verbose(result, verbosity=2):
-            msg += "Result was: %s" % self._dump_results(result._result)
+            msg += "Result was: %s" % self._dump_results(result.result)
         self._display.display(msg, color=C.COLOR_DEBUG)
 
-    def v2_runner_on_async_poll(self, result):
-        host = result._host.get_name()
-        jid = result._result.get('ansible_job_id')
-        started = result._result.get('started')
-        finished = result._result.get('finished')
+    def v2_runner_on_async_poll(self, result: CallbackTaskResult) -> None:
+        host = result.host.get_name()
+        jid = result.result.get('ansible_job_id')
+        started = result.result.get('started')
+        finished = result.result.get('finished')
         self._display.display(
             'ASYNC POLL on %s: jid=%s started=%s finished=%s' % (host, jid, started, finished),
             color=C.COLOR_DEBUG
         )
 
-    def v2_runner_on_async_ok(self, result):
-        host = result._host.get_name()
-        jid = result._result.get('ansible_job_id')
+    def v2_runner_on_async_ok(self, result: CallbackTaskResult) -> None:
+        host = result.host.get_name()
+        jid = result.result.get('ansible_job_id')
         self._display.display("ASYNC OK on %s: jid=%s" % (host, jid), color=C.COLOR_DEBUG)
 
-    def v2_runner_on_async_failed(self, result):
-        host = result._host.get_name()
+    def v2_runner_on_async_failed(self, result: CallbackTaskResult) -> None:
+        host = result.host.get_name()
 
         # Attempt to get the async job ID. If the job does not finish before the
         # async timeout value, the ID may be within the unparsed 'async_result' dict.
-        jid = result._result.get('ansible_job_id')
-        if not jid and 'async_result' in result._result:
-            jid = result._result['async_result'].get('ansible_job_id')
+        jid = result.result.get('ansible_job_id')
+        if not jid and 'async_result' in result.result:
+            jid = result.result['async_result'].get('ansible_job_id')
         self._display.display("ASYNC FAILED on %s: jid=%s" % (host, jid), color=C.COLOR_DEBUG)
 
     def v2_playbook_on_notify(self, handler, host):
diff --git a/lib/ansible/plugins/callback/junit.py b/lib/ansible/plugins/callback/junit.py
index dc56ac5d1b4..812f6fe583a 100644
--- a/lib/ansible/plugins/callback/junit.py
+++ b/lib/ansible/plugins/callback/junit.py
@@ -86,12 +86,14 @@ import decimal
 import os
 import time
 import re
+import typing as t
 
 from ansible import constants
-from ansible.module_utils.common.messages import ErrorSummary
 from ansible.module_utils.common.text.converters import to_bytes, to_text
 from ansible.playbook.task import Task
 from ansible.plugins.callback import CallbackBase
+from ansible.executor.task_result import CallbackTaskResult
+from ansible.playbook.included_file import IncludedFile
 from ansible.utils._junit_xml import (
     TestCase,
     TestError,
@@ -184,23 +186,23 @@ class CallbackModule(CallbackBase):
 
         self._task_data[uuid] = TaskData(uuid, name, path, play, action)
 
-    def _finish_task(self, status, result):
+    def _finish_task(self, status: str, result: IncludedFile | CallbackTaskResult) -> None:
         """ record the results of a task for a single host """
 
-        task_uuid = result._task._uuid
+        if isinstance(result, CallbackTaskResult):
+            task_uuid = result.task._uuid
+            host_uuid = result.host._uuid
+            host_name = result.host.name
 
-        if hasattr(result, '_host'):
-            host_uuid = result._host._uuid
-            host_name = result._host.name
+            if self._fail_on_change == 'true' and status == 'ok' and result.result.get('changed', False):
+                status = 'failed'
         else:
+            task_uuid = result._task._uuid
             host_uuid = 'include'
             host_name = 'include'
 
         task_data = self._task_data[task_uuid]
 
-        if self._fail_on_change == 'true' and status == 'ok' and result._result.get('changed', False):
-            status = 'failed'
-
         # ignore failure if expected and toggle result if asked for
         if status == 'failed' and 'EXPECTED FAILURE' in task_data.name:
             status = 'ok'
@@ -233,7 +235,8 @@ class CallbackModule(CallbackBase):
         if host_data.status == 'included':
             return TestCase(name=name, classname=junit_classname, time=duration, system_out=str(host_data.result))
 
-        res = host_data.result._result
+        task_result = t.cast(CallbackTaskResult, host_data.result)
+        res = task_result.result
         rc = res.get('rc', 0)
         dump = self._dump_results(res, indent=0)
         dump = self._cleanse_string(dump)
@@ -243,10 +246,8 @@ class CallbackModule(CallbackBase):
 
         test_case = TestCase(name=name, classname=junit_classname, time=duration)
 
-        error_summary: ErrorSummary
-
         if host_data.status == 'failed':
-            if error_summary := res.get('exception'):
+            if error_summary := task_result.exception:
                 message = error_summary._format()
                 output = error_summary.formatted_traceback
                 test_case.errors.append(TestError(message=message, output=output))
@@ -309,19 +310,19 @@ class CallbackModule(CallbackBase):
     def v2_playbook_on_handler_task_start(self, task: Task) -> None:
         self._start_task(task)
 
-    def v2_runner_on_failed(self, result, ignore_errors=False):
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors=False) -> None:
         if ignore_errors and self._fail_on_ignore != 'true':
             self._finish_task('ok', result)
         else:
             self._finish_task('failed', result)
 
-    def v2_runner_on_ok(self, result):
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         self._finish_task('ok', result)
 
-    def v2_runner_on_skipped(self, result):
+    def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
         self._finish_task('skipped', result)
 
-    def v2_playbook_on_include(self, included_file):
+    def v2_playbook_on_include(self, included_file: IncludedFile) -> None:
         self._finish_task('included', included_file)
 
     def v2_playbook_on_stats(self, stats):
@@ -347,7 +348,7 @@ class TaskData:
         if host.uuid in self.host_data:
             if host.status == 'included':
                 # concatenate task include output from multiple items
-                host.result = '%s\n%s' % (self.host_data[host.uuid].result, host.result)
+                host.result = f'{self.host_data[host.uuid].result}\n{host.result}'
             else:
                 raise Exception('%s: %s: %s: duplicate host callback: %s' % (self.path, self.play, self.name, host.name))
 
@@ -359,7 +360,7 @@ class HostData:
     Data about an individual host.
     """
 
-    def __init__(self, uuid, name, status, result):
+    def __init__(self, uuid: str, name: str, status: str, result: IncludedFile | CallbackTaskResult | str) -> None:
         self.uuid = uuid
         self.name = name
         self.status = status
diff --git a/lib/ansible/plugins/callback/minimal.py b/lib/ansible/plugins/callback/minimal.py
index 3459a5bc5b5..cff4048bd6f 100644
--- a/lib/ansible/plugins/callback/minimal.py
+++ b/lib/ansible/plugins/callback/minimal.py
@@ -15,7 +15,7 @@ DOCUMENTATION = """
       - result_format_callback
 """
 
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import CallbackTaskResult
 from ansible.plugins.callback import CallbackBase
 from ansible import constants as C
 
@@ -41,41 +41,41 @@ class CallbackModule(CallbackBase):
 
         return buf + "\n"
 
-    def v2_runner_on_failed(self, result: TaskResult, ignore_errors: bool = False) -> None:
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
         self._handle_warnings_and_exception(result)
 
-        if result._task.action in C.MODULE_NO_JSON and 'module_stderr' not in result._result:
-            self._display.display(self._command_generic_msg(result._host.get_name(), result._result, "FAILED"), color=C.COLOR_ERROR)
+        if result.task.action in C.MODULE_NO_JSON and 'module_stderr' not in result.result:
+            self._display.display(self._command_generic_msg(result.host.get_name(), result.result, "FAILED"), color=C.COLOR_ERROR)
         else:
-            self._display.display("%s | FAILED! => %s" % (result._host.get_name(), self._dump_results(result._result, indent=4)), color=C.COLOR_ERROR)
+            self._display.display("%s | FAILED! => %s" % (result.host.get_name(), self._dump_results(result.result, indent=4)), color=C.COLOR_ERROR)
 
-    def v2_runner_on_ok(self, result: TaskResult) -> None:
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         self._handle_warnings_and_exception(result)
 
-        self._clean_results(result._result, result._task.action)
+        self._clean_results(result.result, result.task.action)
 
-        if result._result.get('changed', False):
+        if result.result.get('changed', False):
             color = C.COLOR_CHANGED
             state = 'CHANGED'
         else:
             color = C.COLOR_OK
             state = 'SUCCESS'
 
-        if result._task.action in C.MODULE_NO_JSON and 'ansible_job_id' not in result._result:
-            self._display.display(self._command_generic_msg(result._host.get_name(), result._result, state), color=color)
+        if result.task.action in C.MODULE_NO_JSON and 'ansible_job_id' not in result.result:
+            self._display.display(self._command_generic_msg(result.host.get_name(), result.result, state), color=color)
         else:
-            self._display.display("%s | %s => %s" % (result._host.get_name(), state, self._dump_results(result._result, indent=4)), color=color)
+            self._display.display("%s | %s => %s" % (result.host.get_name(), state, self._dump_results(result.result, indent=4)), color=color)
 
-    def v2_runner_on_skipped(self, result: TaskResult) -> None:
+    def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
         self._handle_warnings_and_exception(result)
 
-        self._display.display("%s | SKIPPED" % (result._host.get_name()), color=C.COLOR_SKIP)
+        self._display.display("%s | SKIPPED" % (result.host.get_name()), color=C.COLOR_SKIP)
 
-    def v2_runner_on_unreachable(self, result: TaskResult) -> None:
+    def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
         self._handle_warnings_and_exception(result)
 
-        self._display.display("%s | UNREACHABLE! => %s" % (result._host.get_name(), self._dump_results(result._result, indent=4)), color=C.COLOR_UNREACHABLE)
+        self._display.display("%s | UNREACHABLE! => %s" % (result.host.get_name(), self._dump_results(result.result, indent=4)), color=C.COLOR_UNREACHABLE)
 
     def v2_on_file_diff(self, result):
-        if 'diff' in result._result and result._result['diff']:
-            self._display.display(self._get_diff(result._result['diff']))
+        if 'diff' in result.result and result.result['diff']:
+            self._display.display(self._get_diff(result.result['diff']))
diff --git a/lib/ansible/plugins/callback/oneline.py b/lib/ansible/plugins/callback/oneline.py
index f5292bae859..7320e45288a 100644
--- a/lib/ansible/plugins/callback/oneline.py
+++ b/lib/ansible/plugins/callback/oneline.py
@@ -16,6 +16,7 @@ DOCUMENTATION = """
 from ansible import constants as C
 from ansible.plugins.callback import CallbackBase
 from ansible.template import Templar
+from ansible.executor.task_result import CallbackTaskResult
 
 
 class CallbackModule(CallbackBase):
@@ -41,9 +42,9 @@ class CallbackModule(CallbackBase):
         else:
             return "%s | %s | rc=%s | (stdout) %s" % (hostname, caption, result.get('rc', -1), stdout)
 
-    def v2_runner_on_failed(self, result, ignore_errors=False):
-        if 'exception' in result._result:
-            error_text = Templar().template(result._result['exception'])  # transform to a string
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
+        if 'exception' in result.result:
+            error_text = Templar().template(result.result['exception'])  # transform to a string
             if self._display.verbosity < 3:
                 # extract just the actual error message from the exception text
                 error = error_text.strip().split('\n')[-1]
@@ -51,31 +52,31 @@ class CallbackModule(CallbackBase):
             else:
                 msg = "An exception occurred during task execution. The full traceback is:\n" + error_text.replace('\n', '')
 
-            if result._task.action in C.MODULE_NO_JSON and 'module_stderr' not in result._result:
-                self._display.display(self._command_generic_msg(result._host.get_name(), result._result, 'FAILED'), color=C.COLOR_ERROR)
+            if result.task.action in C.MODULE_NO_JSON and 'module_stderr' not in result.result:
+                self._display.display(self._command_generic_msg(result.host.get_name(), result.result, 'FAILED'), color=C.COLOR_ERROR)
             else:
                 self._display.display(msg, color=C.COLOR_ERROR)
 
-        self._display.display("%s | FAILED! => %s" % (result._host.get_name(), self._dump_results(result._result, indent=0).replace('\n', '')),
+        self._display.display("%s | FAILED! => %s" % (result.host.get_name(), self._dump_results(result.result, indent=0).replace('\n', '')),
                               color=C.COLOR_ERROR)
 
-    def v2_runner_on_ok(self, result):
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
 
-        if result._result.get('changed', False):
+        if result.result.get('changed', False):
             color = C.COLOR_CHANGED
             state = 'CHANGED'
         else:
             color = C.COLOR_OK
             state = 'SUCCESS'
 
-        if result._task.action in C.MODULE_NO_JSON and 'ansible_job_id' not in result._result:
-            self._display.display(self._command_generic_msg(result._host.get_name(), result._result, state), color=color)
+        if result.task.action in C.MODULE_NO_JSON and 'ansible_job_id' not in result.result:
+            self._display.display(self._command_generic_msg(result.host.get_name(), result.result, state), color=color)
         else:
-            self._display.display("%s | %s => %s" % (result._host.get_name(), state, self._dump_results(result._result, indent=0).replace('\n', '')),
+            self._display.display("%s | %s => %s" % (result.host.get_name(), state, self._dump_results(result.result, indent=0).replace('\n', '')),
                                   color=color)
 
-    def v2_runner_on_unreachable(self, result):
-        self._display.display("%s | UNREACHABLE!: %s" % (result._host.get_name(), result._result.get('msg', '')), color=C.COLOR_UNREACHABLE)
+    def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
+        self._display.display("%s | UNREACHABLE!: %s" % (result.host.get_name(), result.result.get('msg', '')), color=C.COLOR_UNREACHABLE)
 
-    def v2_runner_on_skipped(self, result):
-        self._display.display("%s | SKIPPED" % (result._host.get_name()), color=C.COLOR_SKIP)
+    def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
+        self._display.display("%s | SKIPPED" % (result.host.get_name()), color=C.COLOR_SKIP)
diff --git a/lib/ansible/plugins/callback/tree.py b/lib/ansible/plugins/callback/tree.py
index c67d6cbb817..3d69e8ac96f 100644
--- a/lib/ansible/plugins/callback/tree.py
+++ b/lib/ansible/plugins/callback/tree.py
@@ -30,6 +30,7 @@ DOCUMENTATION = """
 import os
 
 from ansible.constants import TREE_DIR
+from ansible.executor.task_result import CallbackTaskResult
 from ansible.module_utils.common.text.converters import to_bytes, to_text
 from ansible.plugins.callback import CallbackBase
 from ansible.utils.path import makedirs_safe, unfrackpath
@@ -76,14 +77,14 @@ class CallbackModule(CallbackBase):
         except (OSError, IOError) as e:
             self._display.warning(u"Unable to write to %s's file: %s" % (hostname, to_text(e)))
 
-    def result_to_tree(self, result):
-        self.write_tree_file(result._host.get_name(), self._dump_results(result._result))
+    def result_to_tree(self, result: CallbackTaskResult) -> None:
+        self.write_tree_file(result.host.get_name(), self._dump_results(result.result))
 
-    def v2_runner_on_ok(self, result):
+    def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         self.result_to_tree(result)
 
-    def v2_runner_on_failed(self, result, ignore_errors=False):
+    def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
         self.result_to_tree(result)
 
-    def v2_runner_on_unreachable(self, result):
+    def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
         self.result_to_tree(result)
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index ed5b2f1102a..b117604d7d8 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -26,6 +26,7 @@ import sys
 import threading
 import time
 import typing as t
+import collections.abc as _c
 
 from collections import deque
 
@@ -35,12 +36,13 @@ from ansible import context
 from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleParserError, AnsibleTemplateError
 from ansible.executor.play_iterator import IteratingStates, PlayIterator
 from ansible.executor.process.worker import WorkerProcess
-from ansible.executor.task_result import TaskResult, ThinTaskResult
+from ansible.executor.task_result import _RawTaskResult, _WireTaskResult
 from ansible.executor.task_queue_manager import CallbackSend, DisplaySend, PromptSend, TaskQueueManager
 from ansible.module_utils.common.text.converters import to_text
 from ansible.module_utils.connection import Connection, ConnectionError
 from ansible.playbook.handler import Handler
 from ansible.playbook.helpers import load_list_of_blocks
+from ansible.playbook.included_file import IncludedFile
 from ansible.playbook.task import Task
 from ansible.playbook.task_include import TaskInclude
 from ansible.plugins import loader as plugin_loader
@@ -100,10 +102,10 @@ def results_thread_main(strategy: StrategyBase) -> None:
                 dmethod = getattr(display, result.method)
                 dmethod(*result.args, **result.kwargs)
             elif isinstance(result, CallbackSend):
-                task_result = strategy.convert_thin_task_result(result.thin_task_result)
+                task_result = strategy._convert_wire_task_result_to_raw(result.wire_task_result)
                 strategy._tqm.send_callback(result.method_name, task_result)
-            elif isinstance(result, ThinTaskResult):
-                result = strategy.convert_thin_task_result(result)
+            elif isinstance(result, _WireTaskResult):
+                result = strategy._convert_wire_task_result_to_raw(result)
                 with strategy._results_lock:
                     strategy._results.append(result)
             elif isinstance(result, PromptSend):
@@ -135,7 +137,7 @@ def results_thread_main(strategy: StrategyBase) -> None:
 def debug_closure(func):
     """Closure to wrap ``StrategyBase._process_pending_results`` and invoke the task debugger"""
     @functools.wraps(func)
-    def inner(self, iterator, one_pass=False, max_passes=None):
+    def inner(self, iterator: PlayIterator, one_pass: bool = False, max_passes: int | None = None) -> list[_RawTaskResult]:
         status_to_stats_map = (
             ('is_failed', 'failures'),
             ('is_unreachable', 'dark'),
@@ -146,12 +148,12 @@ def debug_closure(func):
         # We don't know the host yet, copy the previous states, for lookup after we process new results
         prev_host_states = iterator.host_states.copy()
 
-        results = func(self, iterator, one_pass=one_pass, max_passes=max_passes)
-        _processed_results = []
+        results: list[_RawTaskResult] = func(self, iterator, one_pass=one_pass, max_passes=max_passes)
+        _processed_results: list[_RawTaskResult] = []
 
         for result in results:
-            task = result._task
-            host = result._host
+            task = result.task
+            host = result.host
             _queued_task_args = self._queued_task_cache.pop((host.name, task._uuid), None)
             task_vars = _queued_task_args['task_vars']
             play_context = _queued_task_args['play_context']
@@ -237,7 +239,7 @@ class StrategyBase:
         # outstanding tasks still in queue
         self._blocked_hosts: dict[str, bool] = dict()
 
-        self._results: deque[TaskResult] = deque()
+        self._results: deque[_RawTaskResult] = deque()
         self._results_lock = threading.Condition(threading.Lock())
 
         # create the result processing thread for reading results in the background
@@ -247,7 +249,7 @@ class StrategyBase:
 
         # holds the list of active (persistent) connections to be shutdown at
         # play completion
-        self._active_connections: dict[str, str] = dict()
+        self._active_connections: dict[Host, str] = dict()
 
         # Caches for get_host calls, to avoid calling excessively
         # These values should be set at the top of the ``run`` method of each
@@ -445,10 +447,10 @@ class StrategyBase:
             for target_host in host_list:
                 _set_host_facts(target_host, always_facts)
 
-    def convert_thin_task_result(self, thin_task_result: ThinTaskResult) -> TaskResult:
-        """Return a `TaskResult` created from a `ThinTaskResult`."""
-        host = self._inventory.get_host(thin_task_result.host_name)
-        queue_cache_entry = (host.name, thin_task_result.task_uuid)
+    def _convert_wire_task_result_to_raw(self, wire_task_result: _WireTaskResult) -> _RawTaskResult:
+        """Return a `_RawTaskResult` created from a `_WireTaskResult`."""
+        host = self._inventory.get_host(wire_task_result.host_name)
+        queue_cache_entry = (host.name, wire_task_result.task_uuid)
 
         try:
             found_task = self._queued_task_cache[queue_cache_entry]['task']
@@ -456,7 +458,7 @@ class StrategyBase:
             # This should only happen due to an implicit task created by the
             # TaskExecutor, restrict this behavior to the explicit use case
             # of an implicit async_status task
-            if thin_task_result.task_fields.get('action') != 'async_status':
+            if wire_task_result.task_fields.get('action') != 'async_status':
                 raise
 
             task = Task()
@@ -464,13 +466,13 @@ class StrategyBase:
             task = found_task.copy(exclude_parent=True, exclude_tasks=True)
             task._parent = found_task._parent
 
-        task.from_attrs(thin_task_result.task_fields)
+        task.from_attrs(wire_task_result.task_fields)
 
-        return TaskResult(
+        return _RawTaskResult(
             host=host,
             task=task,
-            return_data=thin_task_result.return_data,
-            task_fields=thin_task_result.task_fields,
+            return_data=wire_task_result.return_data,
+            task_fields=wire_task_result.task_fields,
         )
 
     def search_handlers_by_notification(self, notification: str, iterator: PlayIterator) -> t.Generator[Handler, None, None]:
@@ -529,7 +531,7 @@ class StrategyBase:
                 yield handler
 
     @debug_closure
-    def _process_pending_results(self, iterator: PlayIterator, one_pass: bool = False, max_passes: int | None = None) -> list[TaskResult]:
+    def _process_pending_results(self, iterator: PlayIterator, one_pass: bool = False, max_passes: int | None = None) -> list[_RawTaskResult]:
         """
         Reads results off the final queue and takes appropriate action
         based on the result (executing callbacks, updating state, etc.).
@@ -545,8 +547,8 @@ class StrategyBase:
             finally:
                 self._results_lock.release()
 
-            original_host = task_result._host
-            original_task: Task = task_result._task
+            original_host = task_result.host
+            original_task: Task = task_result.task
 
             # all host status messages contain 2 entries: (msg, task_result)
             role_ran = False
@@ -580,7 +582,7 @@ class StrategyBase:
                             original_host.name,
                             dict(
                                 ansible_failed_task=original_task.serialize(),
-                                ansible_failed_result=task_result._result,
+                                ansible_failed_result=task_result._return_data,
                             ),
                         )
                     else:
@@ -588,7 +590,7 @@ class StrategyBase:
                 else:
                     self._tqm._stats.increment('ok', original_host.name)
                     self._tqm._stats.increment('ignored', original_host.name)
-                    if 'changed' in task_result._result and task_result._result['changed']:
+                    if task_result.is_changed():
                         self._tqm._stats.increment('changed', original_host.name)
                 self._tqm.send_callback('v2_runner_on_failed', task_result, ignore_errors=ignore_errors)
             elif task_result.is_unreachable():
@@ -610,9 +612,9 @@ class StrategyBase:
                 if original_task.loop:
                     # this task had a loop, and has more than one result, so
                     # loop over all of them instead of a single result
-                    result_items = task_result._result.get('results', [])
+                    result_items = task_result._loop_results
                 else:
-                    result_items = [task_result._result]
+                    result_items = [task_result._return_data]
 
                 for result_item in result_items:
                     if '_ansible_notify' in result_item and task_result.is_changed():
@@ -657,7 +659,7 @@ class StrategyBase:
 
                     if 'add_host' in result_item or 'add_group' in result_item:
                         item_vars = _get_item_vars(result_item, original_task)
-                        found_task_vars = self._queued_task_cache.get((original_host.name, task_result._task._uuid))['task_vars']
+                        found_task_vars = self._queued_task_cache.get((original_host.name, task_result.task._uuid))['task_vars']
                         if item_vars:
                             all_task_vars = combine_vars(found_task_vars, item_vars)
                         else:
@@ -672,17 +674,17 @@ class StrategyBase:
                                 original_task._resolve_conditional(original_task.failed_when, all_task_vars))
 
                         if original_task.loop or original_task.loop_with:
-                            new_item_result = TaskResult(
-                                task_result._host,
-                                task_result._task,
+                            new_item_result = _RawTaskResult(
+                                task_result.host,
+                                task_result.task,
                                 result_item,
-                                task_result._task_fields,
+                                task_result.task_fields,
                             )
                             self._tqm.send_callback('v2_runner_item_on_ok', new_item_result)
                             if result_item.get('changed', False):
-                                task_result._result['changed'] = True
+                                task_result._return_data['changed'] = True
                             if result_item.get('failed', False):
-                                task_result._result['failed'] = True
+                                task_result._return_data['failed'] = True
 
                     if 'ansible_facts' in result_item and original_task.action not in C._ACTION_DEBUG:
                         # if delegated fact and we are delegating facts, we need to change target host for them
@@ -730,13 +732,13 @@ class StrategyBase:
                                 else:
                                     self._tqm._stats.set_custom_stats(k, data[k], myhost)
 
-                if 'diff' in task_result._result:
+                if 'diff' in task_result._return_data:
                     if self._diff or getattr(original_task, 'diff', False):
                         self._tqm.send_callback('v2_on_file_diff', task_result)
 
                 if not isinstance(original_task, TaskInclude):
                     self._tqm._stats.increment('ok', original_host.name)
-                    if 'changed' in task_result._result and task_result._result['changed']:
+                    if task_result.is_changed():
                         self._tqm._stats.increment('changed', original_host.name)
 
                 # finally, send the ok for this task
@@ -746,7 +748,7 @@ class StrategyBase:
             if original_task.register:
                 host_list = self.get_task_hosts(iterator, original_host, original_task)
 
-                clean_copy = strip_internal_keys(module_response_deepcopy(task_result._result))
+                clean_copy = strip_internal_keys(module_response_deepcopy(task_result._return_data))
                 if 'invocation' in clean_copy:
                     del clean_copy['invocation']
 
@@ -797,7 +799,7 @@ class StrategyBase:
 
         return ret_results
 
-    def _copy_included_file(self, included_file):
+    def _copy_included_file(self, included_file: IncludedFile) -> IncludedFile:
         """
         A proven safe and performant way to create a copy of an included file
         """
@@ -810,7 +812,7 @@ class StrategyBase:
 
         return ti_copy
 
-    def _load_included_file(self, included_file, iterator, is_handler=False, handle_stats_and_callbacks=True):
+    def _load_included_file(self, included_file: IncludedFile, iterator, is_handler=False, handle_stats_and_callbacks=True):
         """
         Loads an included YAML file of tasks, applying the optional set of variables.
 
@@ -857,11 +859,11 @@ class StrategyBase:
             else:
                 reason = to_text(e)
             if handle_stats_and_callbacks:
-                for r in included_file._results:
-                    r._result['failed'] = True
+                for tr in included_file._results:
+                    tr._return_data['failed'] = True
 
                 for host in included_file._hosts:
-                    tr = TaskResult(host=host, task=included_file._task, return_data=dict(failed=True, reason=reason), task_fields={})
+                    tr = _RawTaskResult(host=host, task=included_file._task, return_data=dict(failed=True, reason=reason), task_fields={})
                     self._tqm._stats.increment('failures', host.name)
                     self._tqm.send_callback('v2_runner_on_failed', tr)
             raise AnsibleError(reason) from e
@@ -897,7 +899,7 @@ class StrategyBase:
     def _cond_not_supported_warn(self, task_name):
         display.warning("%s task does not support when conditional" % task_name)
 
-    def _execute_meta(self, task: Task, play_context, iterator, target_host):
+    def _execute_meta(self, task: Task, play_context, iterator, target_host: Host):
         task.resolved_action = 'ansible.builtin.meta'  # _post_validate_args is never called for meta actions, so resolved_action hasn't been set
 
         # meta tasks store their args in the _raw_params field of args,
@@ -1075,7 +1077,7 @@ class StrategyBase:
         else:
             display.vv(f"META: {header}")
 
-        res = TaskResult(target_host, task, result, {})
+        res = _RawTaskResult(target_host, task, result, {})
         if skipped:
             self._tqm.send_callback('v2_runner_on_skipped', res)
         return [res]
@@ -1095,14 +1097,14 @@ class StrategyBase:
                     hosts_left.append(self._inventory.get_host(host))
         return hosts_left
 
-    def update_active_connections(self, results):
+    def update_active_connections(self, results: _c.Iterable[_RawTaskResult]) -> None:
         """ updates the current active persistent connections """
         for r in results:
-            if 'args' in r._task_fields:
-                socket_path = r._task_fields['args'].get('_ansible_socket')
+            if 'args' in r.task_fields:
+                socket_path = r.task_fields['args'].get('_ansible_socket')
                 if socket_path:
-                    if r._host not in self._active_connections:
-                        self._active_connections[r._host] = socket_path
+                    if r.host not in self._active_connections:
+                        self._active_connections[r.host] = socket_path
 
 
 class NextAction(object):
diff --git a/lib/ansible/plugins/strategy/free.py b/lib/ansible/plugins/strategy/free.py
index 59227322a4c..f0e4bfe61d0 100644
--- a/lib/ansible/plugins/strategy/free.py
+++ b/lib/ansible/plugins/strategy/free.py
@@ -252,11 +252,11 @@ class StrategyModule(StrategyBase):
                         # FIXME: send the error to the callback; don't directly write to display here
                         display.error(ex)
                         for r in included_file._results:
-                            r._result['failed'] = True
-                            r._result['reason'] = str(ex)
-                            self._tqm._stats.increment('failures', r._host.name)
+                            r._return_data['failed'] = True
+                            r._return_data['reason'] = str(ex)
+                            self._tqm._stats.increment('failures', r.host.name)
                             self._tqm.send_callback('v2_runner_on_failed', r)
-                            failed_includes_hosts.add(r._host)
+                            failed_includes_hosts.add(r.host)
                         continue
                     else:
                         # since we skip incrementing the stats when the task result is
diff --git a/lib/ansible/plugins/strategy/linear.py b/lib/ansible/plugins/strategy/linear.py
index be95d61766f..a9a579efb08 100644
--- a/lib/ansible/plugins/strategy/linear.py
+++ b/lib/ansible/plugins/strategy/linear.py
@@ -40,6 +40,8 @@ from ansible.utils.display import Display
 from ansible.inventory.host import Host
 from ansible.playbook.task import Task
 from ansible.executor.play_iterator import PlayIterator
+from ansible.playbook.play_context import PlayContext
+from ansible.executor import task_result as _task_result
 
 display = Display()
 
@@ -92,7 +94,7 @@ class StrategyModule(StrategyBase):
 
         return host_tasks
 
-    def run(self, iterator, play_context):
+    def run(self, iterator, play_context: PlayContext):  # type: ignore[override]
         """
         The linear strategy is simple - get the next task and queue
         it for all hosts, then wait for the queue to drain before
@@ -100,7 +102,7 @@ class StrategyModule(StrategyBase):
         """
 
         # iterate over each task, while there is one left to run
-        result = self._tqm.RUN_OK
+        result = int(self._tqm.RUN_OK)
         work_to_do = True
 
         self._set_hosts_cache(iterator._play)
@@ -125,7 +127,7 @@ class StrategyModule(StrategyBase):
                 # flag set if task is set to any_errors_fatal
                 any_errors_fatal = False
 
-                results = []
+                results: list[_task_result._RawTaskResult] = []
                 for (host, task) in host_tasks:
                     if self._tqm._terminated:
                         break
@@ -285,11 +287,11 @@ class StrategyModule(StrategyBase):
                             # FIXME: send the error to the callback; don't directly write to display here
                             display.error(ex)
                             for r in included_file._results:
-                                r._result['failed'] = True
-                                r._result['reason'] = str(ex)
-                                self._tqm._stats.increment('failures', r._host.name)
+                                r._return_data['failed'] = True
+                                r._return_data['reason'] = str(ex)
+                                self._tqm._stats.increment('failures', r.host.name)
                                 self._tqm.send_callback('v2_runner_on_failed', r)
-                                failed_includes_hosts.add(r._host)
+                                failed_includes_hosts.add(r.host)
                         else:
                             # since we skip incrementing the stats when the task result is
                             # first processed, we do so now for each host in the list
@@ -320,9 +322,9 @@ class StrategyModule(StrategyBase):
                 unreachable_hosts = []
                 for res in results:
                     if res.is_failed():
-                        failed_hosts.append(res._host.name)
+                        failed_hosts.append(res.host.name)
                     elif res.is_unreachable():
-                        unreachable_hosts.append(res._host.name)
+                        unreachable_hosts.append(res.host.name)
 
                 if any_errors_fatal and (failed_hosts or unreachable_hosts):
                     for host in hosts_left:
diff --git a/lib/ansible/utils/vars.py b/lib/ansible/utils/vars.py
index 59b8cacc7ce..939831f613c 100644
--- a/lib/ansible/utils/vars.py
+++ b/lib/ansible/utils/vars.py
@@ -27,6 +27,7 @@ from json import dumps
 
 from ansible import constants as C
 from ansible import context
+from ansible._internal import _json
 from ansible.errors import AnsibleError, AnsibleOptionsError
 from ansible.module_utils.datatag import native_type_name
 from ansible.module_utils.common.text.converters import to_native, to_text
@@ -284,3 +285,25 @@ def validate_variable_name(name: object) -> None:
         help_text='Variable names must be strings starting with a letter or underscore character, and contain only letters, numbers and underscores.',
         obj=name,
     )
+
+
+def transform_to_native_types(
+    value: object,
+    redact: bool = True,
+) -> object:
+    """
+    Recursively transform the given value to Python native types.
+    Potentially sensitive values such as individually vaulted variables will be redacted unless ``redact=False`` is passed.
+    Which values are considered potentially sensitive may change in future releases.
+    Types which cannot be converted to Python native types will result in an error.
+    """
+    avv = _json.AnsibleVariableVisitor(
+        convert_mapping_to_dict=True,
+        convert_sequence_to_list=True,
+        convert_custom_scalars=True,
+        convert_to_native_values=True,
+        apply_transforms=True,
+        encrypted_string_behavior=_json.EncryptedStringBehavior.REDACT if redact else _json.EncryptedStringBehavior.DECRYPT,
+    )
+
+    return avv.visit(value)
diff --git a/lib/ansible/vars/clean.py b/lib/ansible/vars/clean.py
index 83b840b5492..8be64b2679a 100644
--- a/lib/ansible/vars/clean.py
+++ b/lib/ansible/vars/clean.py
@@ -42,7 +42,7 @@ def module_response_deepcopy(v):
     backwards compatibility, in case we need to extend this function
     to handle our specific needs:
 
-    * ``ansible.executor.task_result.TaskResult.clean_copy``
+    * ``ansible.executor.task_result._RawTaskResult.as_callback_task_result``
     * ``ansible.vars.clean.clean_facts``
     * ``ansible.vars.namespace_facts``
     """
diff --git a/test/integration/targets/callback-legacy-warnings/aliases b/test/integration/targets/callback-legacy-warnings/aliases
new file mode 100644
index 00000000000..5a47349ed9f
--- /dev/null
+++ b/test/integration/targets/callback-legacy-warnings/aliases
@@ -0,0 +1,2 @@
+context/controller
+shippable/posix/group3
diff --git a/test/integration/targets/callback-legacy-warnings/callback_plugins/legacy_warning_display.py b/test/integration/targets/callback-legacy-warnings/callback_plugins/legacy_warning_display.py
new file mode 100644
index 00000000000..c40f41a2a4e
--- /dev/null
+++ b/test/integration/targets/callback-legacy-warnings/callback_plugins/legacy_warning_display.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import collections.abc as c
+import functools
+
+from unittest.mock import MagicMock
+
+from ansible.executor.task_result import CallbackTaskResult
+from ansible.plugins.callback import CallbackBase
+
+
+class CallbackModule(CallbackBase):
+    # DTFIX-RELEASE: validate VaultedValue redaction behavior
+
+    CALLBACK_NEEDS_ENABLED = True
+    seen_tr = []  # track taskresult instances to ensure every call sees a unique instance
+
+    expects_task_result = {
+        'v2_runner_on_failed', 'v2_runner_on_ok', 'v2_runner_on_skipped', 'v2_runner_on_unreachable', 'v2_runner_on_async_poll', 'v2_runner_on_async_ok',
+        'v2_runner_on_async_failed,', 'v2_playbook_on_import_for_host', 'v2_playbook_on_not_import_for_host', 'v2_on_file_diff', 'v2_runner_item_on_ok',
+        'v2_runner_item_on_failed', 'v2_runner_item_on_skipped', 'v2_runner_retry',
+    }
+
+    expects_no_task_result = {
+        'v2_playbook_on_start', 'v2_playbook_on_notify', 'v2_playbook_on_no_hosts_matched', 'v2_playbook_on_no_hosts_remaining', 'v2_playbook_on_task_start',
+        'v2_playbook_on_cleanup_task_start', 'v2_playbook_on_handler_task_start', 'v2_playbook_on_vars_prompt', 'v2_playbook_on_play_start',
+        'v2_playbook_on_stats', 'v2_playbook_on_include', 'v2_runner_on_start',
+    }
+
+    # we're abusing runtime assertions to signify failure in this integration test component; ensure they're not disabled by opimizations
+    try:
+        assert False
+    except AssertionError:
+        pass
+    else:
+        raise BaseException("this test does not function when running Python with optimization")
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._display = MagicMock()
+
+    @staticmethod
+    def get_first_task_result(args: c.Sequence) -> CallbackTaskResult | None:
+        """Find the first CallbackTaskResult in posargs, since the signatures are dynamic and we didn't want to use inspect signature binding."""
+        return next((arg for arg in args if isinstance(arg, CallbackTaskResult)), None)
+
+    def v2_on_any(self, *args, **kwargs) -> None:
+        """Standard behavioral test for the v2_on_any callback method."""
+        print(f'hello from v2_on_any {args=} {kwargs=}')
+        if result := self.get_first_task_result(args):
+            assert isinstance(result, CallbackTaskResult)
+
+            assert result is self._current_task_result
+
+            assert result not in self.seen_tr
+
+            self.seen_tr.append(result)
+        else:
+            assert self._current_task_result is None
+
+    def v2_method_expects_task_result(self, *args, method_name: str, **_kwargs) -> None:
+        """Standard behavioral tests for callback methods accepting a task result; wired dynamically."""
+        print(f'hello from {method_name}')
+        result = self.get_first_task_result(args)
+
+        assert result is self._current_task_result
+
+        assert isinstance(result, CallbackTaskResult)
+
+        assert result not in self.seen_tr
+
+        self.seen_tr.append(result)
+
+        has_exception = bool(result.exception)
+        has_warnings = bool(result.warnings)
+        has_deprecations = bool(result.deprecations)
+
+        self._display.reset_mock()
+
+        self._handle_exception(result.result)  # pops exception from transformed dict
+
+        if has_exception:
+            assert 'exception' not in result.result
+            self._display._error.assert_called()
+
+        self._display.reset_mock()
+
+        self._handle_warnings(result.result)  # pops warnings/deprecations from transformed dict
+
+        if has_warnings:
+            assert 'warnings' not in result.result
+            self._display._warning.assert_called()
+
+        if has_deprecations:
+            assert 'deprecations' not in result.result
+            self._display._deprecated.assert_called()
+
+    def v2_method_expects_no_task_result(self, *args, method_name: str, **_kwargs) -> None:
+        """Standard behavioral tests for non-task result callback methods; wired dynamically."""
+        print(f'hello from {method_name}')
+
+        assert self.get_first_task_result(args) is None
+
+        assert self._current_task_result is None
+
+    def __getattribute__(self, item: str) -> object:
+        if item in CallbackModule.expects_task_result:
+            return functools.partial(CallbackModule.v2_method_expects_task_result, self, method_name=item)
+        elif item in CallbackModule.expects_no_task_result:
+            return functools.partial(CallbackModule.v2_method_expects_no_task_result, self, method_name=item)
+        else:
+            return object.__getattribute__(self, item)
diff --git a/test/integration/targets/callback-legacy-warnings/library/noisy.py b/test/integration/targets/callback-legacy-warnings/library/noisy.py
new file mode 100644
index 00000000000..d402a6db036
--- /dev/null
+++ b/test/integration/targets/callback-legacy-warnings/library/noisy.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from ansible.module_utils.basic import AnsibleModule
+
+
+def main() -> None:
+    m = AnsibleModule({})
+    m.warn("This is a warning.")
+    m.deprecate("This is a deprecation.", version='9999.9')
+    m.exit_json()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/test/integration/targets/callback-legacy-warnings/runme.sh b/test/integration/targets/callback-legacy-warnings/runme.sh
new file mode 100755
index 00000000000..6fd9d37da96
--- /dev/null
+++ b/test/integration/targets/callback-legacy-warnings/runme.sh
@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -eux
+
+ANSIBLE_STDOUT_CALLBACK=legacy_warning_display ansible-playbook test.yml "${@}"
diff --git a/test/integration/targets/callback-legacy-warnings/test.yml b/test/integration/targets/callback-legacy-warnings/test.yml
new file mode 100644
index 00000000000..cd5205d4dd6
--- /dev/null
+++ b/test/integration/targets/callback-legacy-warnings/test.yml
@@ -0,0 +1,20 @@
+- hosts: localhost
+  gather_facts: no
+  tasks:
+    - noisy:
+      register: noisyout
+      async: 5
+      poll: 1
+      loop: [1, 2]
+
+    - noisy:
+      async: 5
+      poll: 1
+      register: noisyout
+
+    - debug:
+      when: false
+
+    - debug:
+        var: 1/0
+      ignore_errors: true
diff --git a/test/integration/targets/callback_results/callback_plugins/track_connections.py b/test/integration/targets/callback_results/callback_plugins/track_connections.py
index de033762f1a..10e872017cc 100644
--- a/test/integration/targets/callback_results/callback_plugins/track_connections.py
+++ b/test/integration/targets/callback_results/callback_plugins/track_connections.py
@@ -15,6 +15,7 @@ import json
 from collections import defaultdict
 
 from ansible.plugins.callback import CallbackBase
+from ansible.executor.task_result import CallbackTaskResult
 
 
 class CallbackModule(CallbackBase):
@@ -26,9 +27,9 @@ class CallbackModule(CallbackBase):
         super().__init__(*args, **kwargs)
         self._conntrack = defaultdict(lambda : defaultdict(int))
 
-    def _track(self, result, *args, **kwargs):
-        host = result._host.get_name()
-        task = result._task
+    def _track(self, result: CallbackTaskResult, *args, **kwargs):
+        host = result.host.get_name()
+        task = result.task
 
         self._conntrack[host][task.connection] += 1
 
diff --git a/test/integration/targets/collections/test_task_resolved_plugin/callback_plugins/display_resolved_action.py b/test/integration/targets/collections/test_task_resolved_plugin/callback_plugins/display_resolved_action.py
index 55d5b6eb179..d8f84824a6f 100644
--- a/test/integration/targets/collections/test_task_resolved_plugin/callback_plugins/display_resolved_action.py
+++ b/test/integration/targets/collections/test_task_resolved_plugin/callback_plugins/display_resolved_action.py
@@ -28,7 +28,7 @@ class CallbackModule(CallbackBase):
         self.requested_to_resolved = {}
 
     def v2_runner_on_ok(self, result):
-        self.requested_to_resolved[result._task.action] = result._task.resolved_action
+        self.requested_to_resolved[result.task.action] = result.task.resolved_action
 
     def v2_playbook_on_stats(self, stats):
         for requested, resolved in self.requested_to_resolved.items():
diff --git a/test/integration/targets/module_utils/callback/pure_json.py b/test/integration/targets/module_utils/callback/pure_json.py
index b60c1b77ecc..8621152a149 100644
--- a/test/integration/targets/module_utils/callback/pure_json.py
+++ b/test/integration/targets/module_utils/callback/pure_json.py
@@ -21,10 +21,10 @@ class CallbackModule(CallbackBase):
     CALLBACK_NAME = 'pure_json'
 
     def v2_runner_on_failed(self, result, ignore_errors=False):
-        self._display.display(json.dumps(result._result))
+        self._display.display(json.dumps(result.result))
 
     def v2_runner_on_ok(self, result):
-        self._display.display(json.dumps(result._result))
+        self._display.display(json.dumps(result.result))
 
     def v2_runner_on_skipped(self, result):
-        self._display.display(json.dumps(result._result))
+        self._display.display(json.dumps(result.result))
diff --git a/test/integration/targets/strategy_host_pinned/callback_plugins/callback_host_count.py b/test/integration/targets/strategy_host_pinned/callback_plugins/callback_host_count.py
index 9d371c037f2..db26ab5b163 100644
--- a/test/integration/targets/strategy_host_pinned/callback_plugins/callback_host_count.py
+++ b/test/integration/targets/strategy_host_pinned/callback_plugins/callback_host_count.py
@@ -39,5 +39,5 @@ class CallbackModule(CallbackBase):
         self._display.display(task.name or task.action)
 
     def v2_runner_on_ok(self, result):
-        if result._task.name == "end":
+        if result.task.name == "end":
             self._executing_hosts_counter -= 1
diff --git a/test/units/executor/test_task_queue_manager_callbacks.py b/test/units/executor/test_task_queue_manager_callbacks.py
deleted file mode 100644
index c53a42d81f8..00000000000
--- a/test/units/executor/test_task_queue_manager_callbacks.py
+++ /dev/null
@@ -1,118 +0,0 @@
-# (c) 2016, Steve Kuznetsov <skuznets@redhat.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/>.
-
-from __future__ import annotations
-
-import unittest
-from unittest.mock import MagicMock
-
-from ansible.executor.task_queue_manager import TaskQueueManager
-from ansible.playbook import Playbook
-from ansible.plugins.callback import CallbackBase
-from ansible.utils import context_objects as co
-
-
-class TestTaskQueueManagerCallbacks(unittest.TestCase):
-    def setUp(self):
-        inventory = MagicMock()
-        variable_manager = MagicMock()
-        loader = MagicMock()
-        passwords = []
-
-        # Reset the stored command line args
-        co.GlobalCLIArgs._Singleton__instance = None
-        self._tqm = TaskQueueManager(inventory, variable_manager, loader, passwords)
-        self._playbook = Playbook(loader)
-
-        # we use a MagicMock to register the result of the call we
-        # expect to `v2_playbook_on_call`. We don't mock out the
-        # method since we're testing code that uses `inspect` to
-        # look at that method's argspec and we want to ensure this
-        # test is easy to reason about.
-        self._register = MagicMock()
-
-    def tearDown(self):
-        # Reset the stored command line args
-        co.GlobalCLIArgs._Singleton__instance = None
-
-    def test_task_queue_manager_callbacks_v2_playbook_on_start(self):
-        """
-        Assert that no exceptions are raised when sending a Playbook
-        start callback to a current callback module plugin.
-        """
-        register = self._register
-
-        class CallbackModule(CallbackBase):
-            """
-            This is a callback module with the current
-            method signature for `v2_playbook_on_start`.
-            """
-            CALLBACK_VERSION = 2.0
-            CALLBACK_TYPE = 'notification'
-            CALLBACK_NAME = 'current_module'
-
-            def v2_playbook_on_start(self, playbook):
-                register(self, playbook)
-
-        callback_module = CallbackModule()
-        self._tqm._callback_plugins.append(callback_module)
-        self._tqm.send_callback('v2_playbook_on_start', self._playbook)
-        register.assert_called_once_with(callback_module, self._playbook)
-
-    def test_task_queue_manager_callbacks_v2_playbook_on_start_wrapped(self):
-        """
-        Assert that no exceptions are raised when sending a Playbook
-        start callback to a wrapped current callback module plugin.
-        """
-        register = self._register
-
-        def wrap_callback(func):
-            """
-            This wrapper changes the exposed argument
-            names for a method from the original names
-            to (*args, **kwargs). This is used in order
-            to validate that wrappers which change par-
-            ameter names do not break the TQM callback
-            system.
-
-            :param func: function to decorate
-            :return: decorated function
-            """
-
-            def wrapper(*args, **kwargs):
-                return func(*args, **kwargs)
-
-            return wrapper
-
-        class WrappedCallbackModule(CallbackBase):
-            """
-            This is a callback module with the current
-            method signature for `v2_playbook_on_start`
-            wrapped in order to change the signature.
-            """
-            CALLBACK_VERSION = 2.0
-            CALLBACK_TYPE = 'notification'
-            CALLBACK_NAME = 'current_module'
-
-            @wrap_callback
-            def v2_playbook_on_start(self, playbook):
-                register(self, playbook)
-
-        callback_module = WrappedCallbackModule()
-        self._tqm._callback_plugins.append(callback_module)
-        self._tqm.send_callback('v2_playbook_on_start', self._playbook)
-        register.assert_called_once_with(callback_module, self._playbook)
diff --git a/test/units/executor/test_task_result.py b/test/units/executor/test_task_result.py
index 80945deca77..d29d5dc91bd 100644
--- a/test/units/executor/test_task_result.py
+++ b/test/units/executor/test_task_result.py
@@ -20,37 +20,37 @@ from __future__ import annotations
 import unittest
 from unittest.mock import MagicMock
 
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import _RawTaskResult
 
 
-class TestTaskResult(unittest.TestCase):
+class TestRawTaskResult(unittest.TestCase):
     def test_task_result_basic(self):
         mock_host = MagicMock()
         mock_task = MagicMock()
 
         # test loading a result with a dict
-        tr = TaskResult(mock_host, mock_task, {}, {})
+        tr = _RawTaskResult(mock_host, mock_task, {}, {})
 
     def test_task_result_is_changed(self):
         mock_host = MagicMock()
         mock_task = MagicMock()
 
         # test with no changed in result
-        tr = TaskResult(mock_host, mock_task, {}, {})
+        tr = _RawTaskResult(mock_host, mock_task, {}, {})
         self.assertFalse(tr.is_changed())
 
         # test with changed in the result
-        tr = TaskResult(mock_host, mock_task, dict(changed=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(changed=True), {})
         self.assertTrue(tr.is_changed())
 
         # test with multiple results but none changed
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
         self.assertFalse(tr.is_changed())
 
         # test with multiple results and one changed
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(changed=False), dict(changed=True), dict(some_key=False)]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(changed=False), dict(changed=True), dict(some_key=False)]), {})
         self.assertTrue(tr.is_changed())
 
     def test_task_result_is_skipped(self):
@@ -58,35 +58,35 @@ class TestTaskResult(unittest.TestCase):
         mock_task = MagicMock()
 
         # test with no skipped in result
-        tr = TaskResult(mock_host, mock_task, dict(), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(), {})
         self.assertFalse(tr.is_skipped())
 
         # test with skipped in the result
-        tr = TaskResult(mock_host, mock_task, dict(skipped=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(skipped=True), {})
         self.assertTrue(tr.is_skipped())
 
         # test with multiple results but none skipped
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
         self.assertFalse(tr.is_skipped())
 
         # test with multiple results and one skipped
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(skipped=False), dict(skipped=True), dict(some_key=False)]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(skipped=False), dict(skipped=True), dict(some_key=False)]), {})
         self.assertFalse(tr.is_skipped())
 
         # test with multiple results and all skipped
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(skipped=True), dict(skipped=True), dict(skipped=True)]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(skipped=True), dict(skipped=True), dict(skipped=True)]), {})
         self.assertTrue(tr.is_skipped())
 
         # test with multiple squashed results (list of strings)
         # first with the main result having skipped=False
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=False), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=False), {})
         self.assertFalse(tr.is_skipped())
         # then with the main result having skipped=True
-        tr = TaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=True), {})
         self.assertTrue(tr.is_skipped())
 
     def test_task_result_is_unreachable(self):
@@ -94,21 +94,21 @@ class TestTaskResult(unittest.TestCase):
         mock_task = MagicMock()
 
         # test with no unreachable in result
-        tr = TaskResult(mock_host, mock_task, {}, {})
+        tr = _RawTaskResult(mock_host, mock_task, {}, {})
         self.assertFalse(tr.is_unreachable())
 
         # test with unreachable in the result
-        tr = TaskResult(mock_host, mock_task, dict(unreachable=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(unreachable=True), {})
         self.assertTrue(tr.is_unreachable())
 
         # test with multiple results but none unreachable
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]), {})
         self.assertFalse(tr.is_unreachable())
 
         # test with multiple results and one unreachable
         mock_task.loop = 'foo'
-        tr = TaskResult(mock_host, mock_task, dict(results=[dict(unreachable=False), dict(unreachable=True), dict(some_key=False)]), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(results=[dict(unreachable=False), dict(unreachable=True), dict(some_key=False)]), {})
         self.assertTrue(tr.is_unreachable())
 
     def test_task_result_is_failed(self):
@@ -116,21 +116,21 @@ class TestTaskResult(unittest.TestCase):
         mock_task = MagicMock()
 
         # test with no failed in result
-        tr = TaskResult(mock_host, mock_task, dict(), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(), {})
         self.assertFalse(tr.is_failed())
 
         # test failed result with rc values (should not matter)
-        tr = TaskResult(mock_host, mock_task, dict(rc=0), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(rc=0), {})
         self.assertFalse(tr.is_failed())
-        tr = TaskResult(mock_host, mock_task, dict(rc=1), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(rc=1), {})
         self.assertFalse(tr.is_failed())
 
         # test with failed in result
-        tr = TaskResult(mock_host, mock_task, dict(failed=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(failed=True), {})
         self.assertTrue(tr.is_failed())
 
         # test with failed_when in result
-        tr = TaskResult(mock_host, mock_task, dict(failed_when_result=True), {})
+        tr = _RawTaskResult(mock_host, mock_task, dict(failed_when_result=True), {})
         self.assertTrue(tr.is_failed())
 
     def test_task_result_no_log(self):
@@ -138,16 +138,16 @@ class TestTaskResult(unittest.TestCase):
         mock_task = MagicMock()
 
         # no_log should remove secrets
-        tr = TaskResult(mock_host, mock_task, dict(_ansible_no_log=True, secret='DONTSHOWME'), {})
-        clean = tr.clean_copy()
-        self.assertTrue('secret' not in clean._result)
+        tr = _RawTaskResult(mock_host, mock_task, dict(_ansible_no_log=True, secret='DONTSHOWME'), {})
+        clean = tr.as_callback_task_result()
+        self.assertTrue('secret' not in clean.result)
 
     def test_task_result_no_log_preserve(self):
         mock_host = MagicMock()
         mock_task = MagicMock()
 
         # no_log should not remove preserved keys
-        tr = TaskResult(
+        tr = _RawTaskResult(
             mock_host,
             mock_task,
             dict(
@@ -159,8 +159,8 @@ class TestTaskResult(unittest.TestCase):
             ),
             task_fields={},
         )
-        clean = tr.clean_copy()
-        self.assertTrue('retries' in clean._result)
-        self.assertTrue('attempts' in clean._result)
-        self.assertTrue('changed' in clean._result)
-        self.assertTrue('foo' not in clean._result)
+        clean = tr.as_callback_task_result()
+        self.assertTrue('retries' in clean.result)
+        self.assertTrue('attempts' in clean.result)
+        self.assertTrue('changed' in clean.result)
+        self.assertTrue('foo' not in clean.result)
diff --git a/test/units/playbook/test_included_file.py b/test/units/playbook/test_included_file.py
index 34223d064d8..94976d82d83 100644
--- a/test/units/playbook/test_included_file.py
+++ b/test/units/playbook/test_included_file.py
@@ -118,9 +118,9 @@ def test_process_include_tasks_results(mock_iterator, mock_variable_manager):
     loaded_task = TaskInclude.load(task_ds, task_include=parent_task)
 
     return_data = {'include': 'include_test.yml'}
-    # The task in the TaskResult has to be a TaskInclude so it has a .static attr
-    result1 = task_result.TaskResult(host=host1, task=loaded_task, return_data=return_data, task_fields={})
-    result2 = task_result.TaskResult(host=host2, task=loaded_task, return_data=return_data, task_fields={})
+    # The task in the _RawTaskResult has to be a TaskInclude so it has a .static attr
+    result1 = task_result._RawTaskResult(host=host1, task=loaded_task, return_data=return_data, task_fields={})
+    result2 = task_result._RawTaskResult(host=host2, task=loaded_task, return_data=return_data, task_fields={})
     results = [result1, result2]
 
     fake_loader = DictDataLoader({'include_test.yml': ""})
@@ -151,11 +151,11 @@ def test_process_include_tasks_diff_files(mock_iterator, mock_variable_manager):
     loaded_child_task._play = None
 
     return_data = {'include': 'include_test.yml'}
-    # The task in the TaskResult has to be a TaskInclude so it has a .static attr
-    result1 = task_result.TaskResult(host=host1, task=loaded_task, return_data=return_data, task_fields={})
+    # The task in the _RawTaskResult has to be a TaskInclude so it has a .static attr
+    result1 = task_result._RawTaskResult(host=host1, task=loaded_task, return_data=return_data, task_fields={})
 
     return_data = {'include': 'other_include_test.yml'}
-    result2 = task_result.TaskResult(host=host2, task=loaded_child_task, return_data=return_data, task_fields={})
+    result2 = task_result._RawTaskResult(host=host2, task=loaded_child_task, return_data=return_data, task_fields={})
     results = [result1, result2]
 
     fake_loader = DictDataLoader({'include_test.yml': "",
@@ -192,9 +192,9 @@ def test_process_include_tasks_simulate_free(mock_iterator, mock_variable_manage
     loaded_task2 = TaskInclude.load(task_ds, task_include=parent_task2)
 
     return_data = {'include': 'include_test.yml'}
-    # The task in the TaskResult has to be a TaskInclude so it has a .static attr
-    result1 = task_result.TaskResult(host=host1, task=loaded_task1, return_data=return_data, task_fields={})
-    result2 = task_result.TaskResult(host=host2, task=loaded_task2, return_data=return_data, task_fields={})
+    # The task in the _RawTaskResult has to be a TaskInclude so it has a .static attr
+    result1 = task_result._RawTaskResult(host=host1, task=loaded_task1, return_data=return_data, task_fields={})
+    result2 = task_result._RawTaskResult(host=host2, task=loaded_task2, return_data=return_data, task_fields={})
     results = [result1, result2]
 
     fake_loader = DictDataLoader({'include_test.yml': ""})
@@ -234,8 +234,8 @@ def test_process_include_simulate_free_block_role_tasks(mock_iterator, mock_vari
         """,
     })
 
-    hostname = "testhost1"
-    hostname2 = "testhost2"
+    host1 = Host("testhost1")
+    host2 = Host("testhost2")
 
     role1_ds = {
         'name': 'task1 include',
@@ -281,16 +281,20 @@ def test_process_include_simulate_free_block_role_tasks(mock_iterator, mock_vari
                                      block=parent_block,
                                      loader=fake_loader)
 
-    result1 = task_result.TaskResult(host=hostname,
-                                     task=include_role1,
-                                     return_data=include_role1_ds,
-                                     task_fields={},
-                                     )
-    result2 = task_result.TaskResult(host=hostname2,
-                                     task=include_role2,
-                                     return_data=include_role2_ds,
-                                     task_fields={},
-                                     )
+    result1 = task_result._RawTaskResult(
+        host=host1,
+        task=include_role1,
+        return_data=include_role1_ds,
+        task_fields={},
+    )
+
+    result2 = task_result._RawTaskResult(
+        host=host2,
+        task=include_role2,
+        return_data=include_role2_ds,
+        task_fields={},
+    )
+
     results = [result1, result2]
 
     res = IncludedFile.process_include_results(results,
@@ -305,8 +309,8 @@ def test_process_include_simulate_free_block_role_tasks(mock_iterator, mock_vari
     # with different tasks
     assert res[0]._task != res[1]._task
 
-    assert res[0]._hosts == ['testhost1']
-    assert res[1]._hosts == ['testhost2']
+    assert res[0]._hosts == [host1]
+    assert res[1]._hosts == [host2]
 
     assert res[0]._args == {}
     assert res[1]._args == {}
diff --git a/test/units/plugins/callback/test_callback.py b/test/units/plugins/callback/test_callback.py
index 0956994ba45..c9232a057c3 100644
--- a/test/units/plugins/callback/test_callback.py
+++ b/test/units/plugins/callback/test_callback.py
@@ -25,7 +25,7 @@ import types
 import unittest
 from unittest.mock import MagicMock
 
-from ansible.executor.task_result import TaskResult
+from ansible.executor.task_result import CallbackTaskResult
 from ansible.inventory.host import Host
 from ansible.plugins.callback import CallbackBase
 
@@ -52,13 +52,13 @@ class TestCallback(unittest.TestCase):
         self.assertIs(cb._display, display_mock)
 
     def test_host_label(self):
-        result = TaskResult(host=Host('host1'), task=mock_task, return_data={}, task_fields={})
+        result = CallbackTaskResult(host=Host('host1'), task=mock_task, return_data={}, task_fields={})
 
         self.assertEqual(CallbackBase.host_label(result), 'host1')
 
     def test_host_label_delegated(self):
         mock_task.delegate_to = 'host2'
-        result = TaskResult(
+        result = CallbackTaskResult(
             host=Host('host1'),
             task=mock_task,
             return_data={'_ansible_delegated_vars': {'ansible_host': 'host2'}},