Misc callback fixes/cleanup (#85344)

* Misc callback fixes/cleanup

* Fix v1 callback method dispatch, fully deprecate v1 methods, add missing tests.
* Clean up callback plugin init/setup code, remove redundancies, improve error messaging.
* Remove unused callback method definitions from base class.

Co-authored-by: Matt Clay <matt@mystile.com>

* switch callback bypass to instance-level from class-level

* preserves any instance-level method magic that implementations were using

* add missing handler dispatch entry

* add tests to ensure all methods are covered

---------

Co-authored-by: Matt Clay <matt@mystile.com>
(cherry picked from commit eec57ec396)

changelogs/fragments/callback_base.yml

@@ -0,0 +1,8 @@
+bugfixes:
+  - callback plugins - Callback plugins that do not extend ``ansible.plugins.callback.CallbackBase`` will fail to load with a warning.
+      If the plugin is used as the stdout callback plugin, this will also be a fatal error.
+  - callback plugins - Removed unused methods - runner_on_no_hosts, playbook_on_setup, playbook_on_import_for_host, playbook_on_not_import_for_host,
+      v2_playbook_on_cleanup_task_start, v2_playbook_on_import_for_host, v2_playbook_on_not_import_for_host.
+  - callback plugins - The stdout callback plugin is no longer called twice if it is also in the list of additional callback plugins.
+  - callback plugins - A more descriptive error is now raised if the stdout callback plugin cannot be loaded.
+  - plugin loader - A warning is now emitted for any plugin which fails to load due to a missing base class.

lib/ansible/cli/adhoc.py

@@ -184,7 +184,7 @@ class AdHocCLI(CLI):
                 variable_manager=variable_manager,
                 loader=loader,
                 passwords=passwords,
-                stdout_callback=cb,
+                stdout_callback_name=cb,
                 run_additional_callbacks=C.DEFAULT_LOAD_CALLBACK_PLUGINS,
                 run_tree=run_tree,
                 forks=context.CLIARGS['forks'],

lib/ansible/cli/console.py

@@ -222,7 +222,7 @@ class ConsoleCLI(CLI, cmd.Cmd):
                     variable_manager=self.variable_manager,
                     loader=self.loader,
                     passwords=self.passwords,
-                    stdout_callback=cb,
+                    stdout_callback_name=cb,
                     run_additional_callbacks=C.DEFAULT_LOAD_CALLBACK_PLUGINS,
                     run_tree=False,
                     forks=self.forks,

lib/ansible/executor/task_queue_manager.py

@@ -34,7 +34,6 @@ from ansible.executor.play_iterator import PlayIterator
 from ansible.executor.stats import AggregateStats
 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
 from ansible.parsing.dataloader import DataLoader
 from ansible.playbook.play_context import PlayContext
@@ -139,7 +138,7 @@ class TaskQueueManager:
         variable_manager: VariableManager,
         loader: DataLoader,
         passwords: dict[str, str | None],
-        stdout_callback: str | None = None,
+        stdout_callback_name: str | None = None,
         run_additional_callbacks: bool = True,
         run_tree: bool = False,
         forks: int | None = None,
@@ -149,12 +148,11 @@ class TaskQueueManager:
         self._loader = loader
         self._stats = AggregateStats()
         self.passwords = passwords
-        self._stdout_callback: str | None | CallbackBase = stdout_callback
+        self._stdout_callback_name: str | None = stdout_callback_name or C.DEFAULT_STDOUT_CALLBACK
         self._run_additional_callbacks = run_additional_callbacks
         self._run_tree = run_tree
         self._forks = forks or 5
 
-        self._callbacks_loaded = False
         self._callback_plugins: list[CallbackBase] = []
         self._start_at_done = False
 
@@ -199,44 +197,40 @@ class TaskQueueManager:
         only one such callback plugin will be loaded.
         """
 
-        if self._callbacks_loaded:
+        if self._callback_plugins:
             return
 
-        stdout_callback_loaded = False
+        if not self._stdout_callback_name:
-        if self._stdout_callback is None:
+            raise AnsibleError("No stdout callback name provided.")
-            self._stdout_callback = C.DEFAULT_STDOUT_CALLBACK
 
-        if isinstance(self._stdout_callback, CallbackBase):
+        stdout_callback = callback_loader.get(self._stdout_callback_name)
-            stdout_callback_loaded = True
+
-        elif isinstance(self._stdout_callback, string_types):
+        if not stdout_callback:
-            if self._stdout_callback not in callback_loader:
+            raise AnsibleError(f"Could not load {self._stdout_callback_name!r} callback plugin.")
-                raise AnsibleError("Invalid callback for stdout specified: %s" % self._stdout_callback)
+
-            else:
+        stdout_callback._init_callback_methods()
-                self._stdout_callback = callback_loader.get(self._stdout_callback)
+        stdout_callback.set_options()
-                self._stdout_callback.set_options()
+
-                stdout_callback_loaded = True
+        self._callback_plugins.append(stdout_callback)
-        else:
-            raise AnsibleError("callback must be an instance of CallbackBase or the name of a callback plugin")
 
         # get all configured loadable callbacks (adjacent, builtin)
-        callback_list = list(callback_loader.all(class_only=True))
+        plugin_types = {plugin_type.ansible_name: plugin_type for plugin_type in callback_loader.all(class_only=True)}
 
         # add enabled callbacks that refer to collections, which might not appear in normal listing
         for c in C.CALLBACKS_ENABLED:
             # load all, as collection ones might be using short/redirected names and not a fqcn
             plugin = callback_loader.get(c, class_only=True)
 
-            # TODO: check if this skip is redundant, loader should handle bad file/plugin cases already
             if plugin:
                 # avoids incorrect and dupes possible due to collections
-                if plugin not in callback_list:
+                plugin_types.setdefault(plugin.ansible_name, plugin)
-                    callback_list.append(plugin)
             else:
                 display.warning("Skipping callback plugin '%s', unable to load" % c)
 
-        # for each callback in the list see if we should add it to 'active callbacks' used in the play
+        plugin_types.pop(stdout_callback.ansible_name, None)
-        for callback_plugin in callback_list:
 
+        # for each callback in the list see if we should add it to 'active callbacks' used in the play
+        for callback_plugin in plugin_types.values():
             callback_type = getattr(callback_plugin, 'CALLBACK_TYPE', '')
             callback_needs_enabled = getattr(callback_plugin, 'CALLBACK_NEEDS_ENABLED', getattr(callback_plugin, 'CALLBACK_NEEDS_WHITELIST', False))
 
@@ -252,10 +246,8 @@ class TaskQueueManager:
             display.vvvvv("Attempting to use '%s' callback." % (callback_name))
             if callback_type == 'stdout':
                 # we only allow one callback of type 'stdout' to be loaded,
-                if callback_name != self._stdout_callback or stdout_callback_loaded:
+                display.vv("Skipping callback '%s', as we already have a stdout callback." % (callback_name))
-                    display.vv("Skipping callback '%s', as we already have a stdout callback." % (callback_name))
+                continue
-                    continue
-                stdout_callback_loaded = True
             elif callback_name == 'tree' and self._run_tree:
                 # TODO: remove special case for tree, which is an adhoc cli option --tree
                 pass
@@ -270,21 +262,16 @@ class TaskQueueManager:
                 # avoid bad plugin not returning an object, only needed cause we do class_only load and bypass loader checks,
                 # really a bug in the plugin itself which we ignore as callback errors are not supposed to be fatal.
                 if callback_obj:
-                    # skip initializing if we already did the work for the same plugin (even with diff names)
+                    callback_obj._init_callback_methods()
-                    if callback_obj not in self._callback_plugins:
+                    callback_obj.set_options()
-                        callback_obj.set_options()
+                    self._callback_plugins.append(callback_obj)
-                        self._callback_plugins.append(callback_obj)
-                    else:
-                        display.vv("Skipping callback '%s', already loaded as '%s'." % (callback_plugin, callback_name))
                 else:
                     display.warning("Skipping callback '%s', as it does not create a valid plugin instance." % callback_name)
                     continue
-            except Exception as e:
+            except Exception as ex:
-                display.warning("Skipping callback '%s', unable to load due to: %s" % (callback_name, to_native(e)))
+                display.warning_as_error(f"Failed to load callback plugin {callback_name!r}.", exception=ex)
                 continue
 
-        self._callbacks_loaded = True
-
     def run(self, play):
         """
         Iterates over the roles/tasks in a play, using the given (or default)
@@ -294,8 +281,7 @@ class TaskQueueManager:
         are done with the current task).
         """
 
-        if not self._callbacks_loaded:
+        self.load_callbacks()
-            self.load_callbacks()
 
         all_vars = self._variable_manager.get_vars(play=play)
         templar = TemplateEngine(loader=self._loader, variables=all_vars)
@@ -311,13 +297,9 @@ class TaskQueueManager:
         )
 
         play_context = PlayContext(new_play, self.passwords, self._connection_lockfile.fileno())
-        if (self._stdout_callback and
-                hasattr(self._stdout_callback, 'set_play_context')):
-            self._stdout_callback.set_play_context(play_context)
 
         for callback_plugin in self._callback_plugins:
-            if hasattr(callback_plugin, 'set_play_context'):
+            callback_plugin.set_play_context(play_context)
-                callback_plugin.set_play_context(play_context)
 
         self.send_callback('v2_playbook_on_play_start', new_play)
 
@@ -437,7 +419,7 @@ class TaskQueueManager:
     @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:
+        for callback_plugin in 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 callback_plugin.disabled:
@@ -448,31 +430,13 @@ class TaskQueueManager:
             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']:
+            if method_name in callback_plugin._implemented_callback_methods:
-                method = getattr(callback_plugin, possible, None)
+                methods.append(getattr(callback_plugin, method_name))
-
-                if method is None:
-                    method = getattr(callback_plugin, possible.removeprefix('v2_'), None)
-
-                    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)
+            if 'v2_on_any' in callback_plugin._implemented_callback_methods:
+                methods.append(getattr(callback_plugin, 'v2_on_any'))
 
             for method in methods:
                 # send clean copies
@@ -498,4 +462,4 @@ class TaskQueueManager:
                     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
+            callback_plugin._current_task_result = None  # clear temporary instance storage hack

lib/ansible/module_utils/_internal/_deprecator.py

@@ -38,11 +38,16 @@ def get_caller_plugin_info() -> _messages.PluginInfo | None:
     _skip_stackwalk = True
 
     if frame_info := _stack.caller_frame():
-        return _path_as_core_plugininfo(frame_info.filename) or _path_as_collection_plugininfo(frame_info.filename)
+        return _path_as_plugininfo(frame_info.filename)
 
     return None  # pragma: nocover
 
 
+def _path_as_plugininfo(path: str) -> _messages.PluginInfo | None:
+    """Return a `PluginInfo` instance if the provided `path` refers to a plugin."""
+    return _path_as_core_plugininfo(path) or _path_as_collection_plugininfo(path)
+
+
 def _path_as_core_plugininfo(path: str) -> _messages.PluginInfo | None:
     """Return a `PluginInfo` instance if the provided `path` refers to a core plugin."""
     try:
@@ -62,6 +67,10 @@ def _path_as_core_plugininfo(path: str) -> _messages.PluginInfo | None:
             # Callers in this case need to identify the deprecating plugin name, otherwise only ansible-core will be reported.
             # Reporting ansible-core is never wrong, it just may be missing an additional detail (plugin name) in the "on behalf of" case.
             return ANSIBLE_CORE_DEPRECATOR
+
+        if plugin_name == '__init__':
+            # The plugin type is known, but the caller isn't a specific plugin -- instead, it's core plugin infrastructure (the base class).
+            return _messages.PluginInfo(resolved_name=namespace, type=plugin_type)
     elif match := re.match(r'modules/(?P<module_name>\w+)', relpath):
         # AnsiballZ Python package for core modules
         plugin_name = match.group("module_name")
@@ -101,6 +110,8 @@ def _path_as_collection_plugininfo(path: str) -> _messages.PluginInfo | None:
 
     name = '.'.join((match.group('ns'), match.group('coll'), match.group('plugin_name')))
 
+    # DTFIX-FUTURE: deprecations from __init__ will be incorrectly attributed to a plugin of that name
+
     return _messages.PluginInfo(resolved_name=name, type=plugin_type)
 
 

lib/ansible/plugins/callback/__init__.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import difflib
 import functools
+import inspect
 import json
 import re
 import sys
@@ -39,6 +40,7 @@ from ansible.utils.display import Display
 from ansible.vars.clean import strip_internal_keys, module_response_deepcopy
 from ansible.module_utils._internal._json._profiles import _fallback_to_str
 from ansible._internal._templating import _engine
+from ansible.module_utils._internal import _deprecator
 
 import yaml
 
@@ -61,16 +63,6 @@ _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)
@@ -154,6 +146,9 @@ class CallbackBase(AnsiblePlugin):
     custom actions.
     """
 
+    _implemented_callback_methods: frozenset[str] = frozenset()
+    """Set of callback methods overridden by each subclass; used by TQM to bypass callback dispatch on no-op methods."""
+
     def __init__(self, display: Display | None = None, options: dict[str, t.Any] | None = None) -> None:
         super().__init__()
 
@@ -189,6 +184,48 @@ class CallbackBase(AnsiblePlugin):
     # helper for callbacks, so they don't all have to include deepcopy
     _copy_result = deepcopy
 
+    def _init_callback_methods(self) -> None:
+        """Record analysis of callback methods on each callback instance for dispatch optimization and deprecation warnings."""
+        implemented_callback_methods: set[str] = set()
+        deprecated_v1_method_overrides: set[str] = set()
+        plugin_file = sys.modules[type(self).__module__].__file__
+
+        if plugin_info := _deprecator._path_as_plugininfo(plugin_file):
+            plugin_name = plugin_info.resolved_name
+        else:
+            plugin_name = plugin_file
+
+        for base_v2_method, base_v1_method in CallbackBase._v2_v1_method_map.items():
+            method_name = None
+
+            if not inspect.ismethod(method := getattr(self, (v2_method_name := base_v2_method.__name__))) or method.__func__ is not base_v2_method:
+                implemented_callback_methods.add(v2_method_name)  # v2 method directly implemented by subclass
+                method_name = v2_method_name
+            elif base_v1_method is None:
+                pass  # no corresponding v1 method
+            elif not inspect.ismethod(method := getattr(self, (v1_method_name := base_v1_method.__name__))) or method.__func__ is not base_v1_method:
+                implemented_callback_methods.add(v2_method_name)  # v1 method directly implemented by subclass
+                deprecated_v1_method_overrides.add(v1_method_name)
+                method_name = v1_method_name
+
+            if method_name and v2_method_name == 'v2_on_any':
+                deprecated_v1_method_overrides.discard(method_name)  # avoid including v1 on_any in the v1 deprecation below
+
+                global_display.deprecated(
+                    msg=f'The {plugin_name!r} callback plugin implements deprecated method {method_name!r}.',
+                    version='2.23',
+                    help_text='Use event-specific callback methods instead.',
+                )
+
+        self._implemented_callback_methods = frozenset(implemented_callback_methods)
+
+        if deprecated_v1_method_overrides:
+            global_display.deprecated(
+                msg=f'The {plugin_name!r} callback plugin implements the following deprecated method(s): {", ".join(sorted(deprecated_v1_method_overrides))}',
+                version='2.23',
+                help_text='Implement the `v2_*` equivalent callback method(s) instead.',
+            )
+
     def set_option(self, k, v):
         self._plugin_options[k] = C.config.get_config_value(k, plugin_type=self.plugin_type, plugin_name=self._load_name, direct={k: v})
 
@@ -471,96 +508,61 @@ 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)
 
-    @_callback_base_impl
     def v2_runner_on_failed(self, result: CallbackTaskResult, ignore_errors: bool = False) -> None:
         """Process results of a failed task.
 
@@ -583,7 +585,6 @@ class CallbackBase(AnsiblePlugin):
         host = result.host.get_name()
         self.runner_on_failed(host, result.result, ignore_errors)
 
-    @_callback_base_impl
     def v2_runner_on_ok(self, result: CallbackTaskResult) -> None:
         """Process results of a successful task.
 
@@ -596,7 +597,6 @@ class CallbackBase(AnsiblePlugin):
         host = result.host.get_name()
         self.runner_on_ok(host, result.result)
 
-    @_callback_base_impl
     def v2_runner_on_skipped(self, result: CallbackTaskResult) -> None:
         """Process results of a skipped task.
 
@@ -610,7 +610,6 @@ class CallbackBase(AnsiblePlugin):
             host = result.host.get_name()
             self.runner_on_skipped(host, self._get_item_label(getattr(result.result, 'results', {})))
 
-    @_callback_base_impl
     def v2_runner_on_unreachable(self, result: CallbackTaskResult) -> None:
         """Process results of a task if a target node is unreachable.
 
@@ -623,7 +622,6 @@ class CallbackBase(AnsiblePlugin):
         host = result.host.get_name()
         self.runner_on_unreachable(host, result.result)
 
-    @_callback_base_impl
     def v2_runner_on_async_poll(self, result: CallbackTaskResult) -> None:
         """Get details about an unfinished task running in async mode.
 
@@ -642,7 +640,6 @@ class CallbackBase(AnsiblePlugin):
         clock = 0
         self.runner_on_async_poll(host, result.result, jid, clock)
 
-    @_callback_base_impl
     def v2_runner_on_async_ok(self, result: CallbackTaskResult) -> None:
         """Process results of a successful task that ran in async mode.
 
@@ -656,7 +653,6 @@ class CallbackBase(AnsiblePlugin):
         jid = result.result.get('ansible_job_id')
         self.runner_on_async_ok(host, result.result, jid)
 
-    @_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
@@ -666,89 +662,84 @@ class CallbackBase(AnsiblePlugin):
             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
-    @_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
-    @_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)
 
-    @_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
 
-    @_callback_base_impl
     def v2_runner_item_on_ok(self, result: CallbackTaskResult) -> None:
         pass
 
-    @_callback_base_impl
     def v2_runner_item_on_failed(self, result: CallbackTaskResult) -> None:
         pass
 
-    @_callback_base_impl
     def v2_runner_item_on_skipped(self, result: CallbackTaskResult) -> None:
         pass
 
-    @_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
 
         .. versionadded:: 2.8
         """
         pass
+
+    _v2_v1_method_map = {
+        v2_on_any: on_any,
+        v2_on_file_diff: on_file_diff,
+        v2_playbook_on_handler_task_start: None,
+        v2_playbook_on_include: None,
+        v2_playbook_on_no_hosts_matched: playbook_on_no_hosts_matched,
+        v2_playbook_on_no_hosts_remaining: playbook_on_no_hosts_remaining,
+        v2_playbook_on_notify: playbook_on_notify,
+        v2_playbook_on_play_start: playbook_on_play_start,
+        v2_playbook_on_start: playbook_on_start,
+        v2_playbook_on_stats: playbook_on_stats,
+        v2_playbook_on_task_start: playbook_on_task_start,
+        v2_playbook_on_vars_prompt: playbook_on_vars_prompt,
+        v2_runner_item_on_failed: None,
+        v2_runner_item_on_ok: None,
+        v2_runner_item_on_skipped: None,
+        v2_runner_on_async_failed: runner_on_async_failed,
+        v2_runner_on_async_ok: runner_on_async_ok,
+        v2_runner_on_async_poll: runner_on_async_poll,
+        v2_runner_on_failed: runner_on_failed,
+        v2_runner_on_ok: runner_on_ok,
+        v2_runner_on_skipped: runner_on_skipped,
+        v2_runner_on_start: None,
+        v2_runner_on_unreachable: runner_on_unreachable,
+        v2_runner_retry: None,
+    }
+    """Internal mapping of v2 callback methods with v1 counterparts; populated after type init for deprecation warnings and bypass calculation."""

lib/ansible/plugins/callback/default.py

@@ -197,9 +197,6 @@ class CallbackModule(CallbackBase):
 
         self._last_task_banner = task._uuid
 
-    def v2_playbook_on_cleanup_task_start(self, task):
-        self._task_start(task, prefix='CLEANUP TASK')
-
     def v2_playbook_on_handler_task_start(self, task):
         self._task_start(task, prefix='RUNNING HANDLER')
 

lib/ansible/plugins/callback/junit.py

@@ -300,15 +300,9 @@ class CallbackModule(CallbackBase):
     def v2_playbook_on_play_start(self, play):
         self._play_name = play.get_name()
 
-    def v2_runner_on_no_hosts(self, task: Task) -> None:
-        self._start_task(task)
-
     def v2_playbook_on_task_start(self, task: Task, is_conditional: bool) -> None:
         self._start_task(task)
 
-    def v2_playbook_on_cleanup_task_start(self, task: Task) -> None:
-        self._start_task(task)
-
     def v2_playbook_on_handler_task_start(self, task: Task) -> None:
         self._start_task(task)
 

lib/ansible/plugins/loader.py

@@ -989,6 +989,9 @@ class PluginLoader:
     def get_with_context(self, name, *args, **kwargs) -> get_with_context_result:
         """ instantiates a plugin of the given name using arguments """
 
+        if not name:
+            raise ValueError('A non-empty plugin name is required.')
+
         found_in_cache = True
         class_only = kwargs.pop('class_only', False)
         collection_list = kwargs.pop('collection_list', None)
@@ -1034,6 +1037,7 @@ class PluginLoader:
             except AttributeError:
                 return get_with_context_result(None, plugin_load_context)
             if not issubclass(obj, plugin_class):
+                display.warning(f"Ignoring {self.type} plugin {resolved_type_name!r} due to missing base class {self.base_class!r}.")
                 return get_with_context_result(None, plugin_load_context)
 
         # FIXME: update this to use the load context
@@ -1721,6 +1725,7 @@ callback_loader = PluginLoader(
     'ansible.plugins.callback',
     C.DEFAULT_CALLBACK_PLUGIN_PATH,
     'callback_plugins',
+    required_base_class='CallbackBase',
 )
 
 connection_loader = PluginLoader(

lib/ansible/utils/display.py

@@ -621,6 +621,12 @@ class Display(metaclass=Singleton):
             # collections have a resolved_name but no type
             collection = deprecator.resolved_name if deprecator else None
             plugin_fragment = ''
+        elif deprecator.resolved_name == 'ansible.builtin':
+            # core deprecations from base classes (the API) have no plugin name, only 'ansible.builtin'
+            plugin_type_name = str(deprecator.type) if deprecator.type is _messages.PluginType.MODULE else f'{deprecator.type} plugin'
+
+            collection = deprecator.resolved_name
+            plugin_fragment = f'the {plugin_type_name} API'
         else:
             parts = deprecator.resolved_name.split('.')
             plugin_name = parts[-1]

test/integration/targets/callback-dispatch/callback_plugins/legacy_warning_display.py

@@ -17,14 +17,14 @@ class CallbackModule(CallbackBase):
 
     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_on_async_failed,', '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_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',
+        '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
@@ -103,6 +103,12 @@ class CallbackModule(CallbackBase):
 
         assert self._current_task_result is None
 
+    def v2_playbook_on_stats(self, *args, **kwargs) -> None:
+        print('hello from v2_playbook_on_stats')
+        assert self.get_first_task_result(args) is None
+
+        print('legacy warning display callback test PASS')
+
     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)

test/integration/targets/callback-dispatch/callback_plugins/missing_base_class.py

@@ -0,0 +1,6 @@
+from __future__ import annotations
+
+
+class CallbackModule:
+    """This callback should fail to load since it doesn't extend the required builtin base class."""
+    pass

test/integration/targets/callback-dispatch/callback_plugins/oops_always_enabled.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import os
+import typing as t
+
+from ansible.plugins.callback import CallbackBase
+
+
+class CallbackModule(CallbackBase):
+    call_count: t.ClassVar[int] = 0
+
+    def v2_runner_on_ok(self, *args, **kwargs) -> None:
+        print(f"hello from ALWAYS ENABLED v2_runner_on_ok {args=} {kwargs=}")
+
+        CallbackModule.call_count += 1
+
+    def v2_playbook_on_stats(self, stats):
+        print('hello from ALWAYS ENABLED v2_playbook_on_stats')
+
+        if os.environ.get('_ASSERT_OOPS'):
+            assert CallbackModule.call_count < 2, "always enabled callback should not "
+            print("no double callbacks test PASS")

test/integration/targets/callback-dispatch/callback_plugins/v1_only_methods.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import functools
+
+from ansible.plugins.callback import CallbackBase
+
+
+class CallbackModule(CallbackBase):
+    """Test callback that implements exclusively deprecated v1 callback methods."""
+    CALLBACK_NEEDS_ENABLED = True
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.called_v1_method_names: set[str] = set()
+
+    def callback_impl(self, *args, name: str, **kwargs) -> None:
+        print(f"hi from callback {name!r} with {args=!r} {kwargs=!r}")
+        self.called_v1_method_names.add(name)
+
+    for v1_method in CallbackBase._v2_v1_method_map.values():
+        if not v1_method:
+            continue
+
+        locals()[v1_method.__name__] = functools.partialmethod(callback_impl, name=v1_method.__name__)
+
+    def playbook_on_stats(self, stats, *args, **kwargs):
+        if missed_v1_method_calls := (
+                {'on_any',
+                 'runner_on_ok',
+                 'playbook_on_task_start',
+                 'runner_on_async_ok',
+                 } - self.called_v1_method_names):
+            assert False, f"The following v1 callback methods were not invoked as expected: {', '.join(missed_v1_method_calls)}"
+
+        print("v1 callback test PASS")

test/integration/targets/callback-dispatch/one-task.yml

@@ -0,0 +1,4 @@
+- hosts: localhost
+  gather_facts: no
+  tasks:
+    - debug:

test/integration/targets/callback-dispatch/runme.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+
+set -eux -o pipefail
+
+# the callback itself will raise AssertionError and fail if called > 1x
+_ASSERT_OOPS=1 ANSIBLE_STDOUT_CALLBACK=oops_always_enabled ansible-playbook one-task.yml > no_double_callbacks.txt 2>&1
+
+grep 'no double callbacks test PASS' no_double_callbacks.txt
+
+if ANSIBLE_FORCE_COLOR=0 ANSIBLE_STDOUT_CALLBACK=missing_base_class ansible-playbook one-task.yml > missing_base_class.txt 2>&1; then false; else true; fi
+
+grep "due to missing base class 'CallbackBase'" missing_base_class.txt
+
+# the callback itself will raise AssertionError and fail if some callback methods do not execute
+ANSIBLE_STDOUT_CALLBACK=legacy_warning_display ansible-playbook test_legacy_warning_display.yml "${@}" 2>&1 | tee legacy_warning_out.txt
+
+grep 'legacy warning display callback test PASS' legacy_warning_out.txt
+
+# the callback itself will raise AssertionError and fail if some callback methods do not execute
+ANSIBLE_STDOUT_CALLBACK=v1_only_methods ansible-playbook test_v1_methods.yml "${@}" | tee v1_methods_out.txt
+
+grep 'v1 callback test PASS' v1_methods_out.txt

test/integration/targets/callback-dispatch/test_v1_methods.yml

@@ -0,0 +1,9 @@
+- hosts: localhost
+  gather_facts: no
+  tasks:
+    - debug:
+    - debug:
+      loop: [1]
+    - shell: echo hey
+      async: 2
+      poll: 1

test/integration/targets/callback-legacy-warnings/runme.sh

@@ -1,5 +0,0 @@
-#!/usr/bin/env bash
-
-set -eux
-
-ANSIBLE_STDOUT_CALLBACK=legacy_warning_display ansible-playbook test.yml "${@}"

test/units/plugins/callback/test_callback.py

@@ -413,3 +413,13 @@ class TestCallbackOnMethods(unittest.TestCase):
         cb = CallbackBase()
         cb.v2_on_any('whatever', some_keyword='blippy')
         cb.on_any('whatever', some_keyword='blippy')
+
+
+def test_v2_v1_method_map() -> None:
+    """Ensure that all v2 callback methods appear in the method map."""
+    expected_names = [name for name in dir(CallbackBase) if name.startswith('v2_')]
+    mapped_names = {method.__name__ for method in CallbackBase._v2_v1_method_map}
+
+    missing = [name for name in expected_names if name not in mapped_names]
+
+    assert not missing