feat(implementation): add NormalImpl, WrapperImpl, and CompletionHook setup API

Complete design step 04:

- HookImpl becomes a base class storing hookimpl_config (deprecated
  .opts alias kept) with arg binding moved to _get_call_args.
- NormalImpl / WrapperImpl subclasses validate their configuration;
  HookimplConfiguration.create_hookimpl() returns the right subclass
  (fixing the try-claude footgun of bare HookImpl for normals).
- WrapperImpl.setup_and_get_completion_hook() runs wrapper setup and
  returns a CompletionHook (runtime-checkable Protocol) that owns
  teardown, adapting old-style hookwrappers uniformly.
- Registration and call_extra construct impls via create_hookimpl;
  multicall binds args via _get_call_args. Full dual-sequence multicall
  rewiring lands with design step 05.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: RonnyPfannschmidt

changelog/706.feature.rst

@@ -0,0 +1,9 @@
+Hook implementations are now represented by dedicated types:
+:class:`pluggy.NormalImpl` for normal implementations and
+:class:`pluggy.WrapperImpl` for (old- and new-style) wrappers, both
+subclasses of :class:`pluggy.HookImpl`.
+``HookimplConfiguration.create_hookimpl()`` selects the appropriate
+subclass, and ``WrapperImpl.setup_and_get_completion_hook()`` exposes
+wrapper setup/teardown as a ``CompletionHook`` callback.
+``HookImpl`` now stores its configuration as ``hookimpl_config``; the old
+``opts`` attribute remains as a deprecated alias property.

docs/api_reference.rst

@@ -40,6 +40,14 @@ API Reference
 .. autoclass:: pluggy.HookImpl()
     :members:
 
+.. autoclass:: pluggy.NormalImpl()
+    :show-inheritance:
+    :members:
+
+.. autoclass:: pluggy.WrapperImpl()
+    :show-inheritance:
+    :members:
+
 .. autoclass:: pluggy.HookspecConfiguration()
     :members:
 

src/pluggy/__init__.py

@@ -9,6 +9,8 @@
     "HookspecOpts",
     "HookimplOpts",
     "HookImpl",
+    "NormalImpl",
+    "WrapperImpl",
     "HookRelay",
     "HookspecMarker",
     "HookimplMarker",
@@ -23,6 +25,8 @@
 from ._hooks import HookimplMarker
 from ._hooks import HookRelay
 from ._hooks import HookspecMarker
+from ._hooks import NormalImpl
+from ._hooks import WrapperImpl
 from ._manager import PluginManager
 from ._manager import PluginValidationError
 from ._pytest_compat import HookimplOpts

src/pluggy/_caller.py

@@ -224,10 +224,10 @@ def call_extra(
             "Cannot directly call a historic hook - use call_historic instead."
         )
         self._verify_all_args_are_provided(kwargs)
-        opts = HookimplConfiguration()
+        config = HookimplConfiguration()
         hookimpls = self._hookimpls.copy()
         for method in methods:
-            hookimpl = HookImpl(None, "<temp>", method, opts)
+            hookimpl = config.create_hookimpl(None, "<temp>", method)
             # Find last non-tryfirst nonwrapper method.
             i = len(hookimpls) - 1
             while i >= 0 and (

src/pluggy/_config.py

@@ -8,6 +8,14 @@
 from typing import Any
 from typing import Final
 from typing import final
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from ._implementation import _HookImplFunction
+    from ._implementation import _Plugin
+    from ._implementation import NormalImpl
+    from ._implementation import WrapperImpl
 
 
 @final
@@ -98,6 +106,27 @@ def __init__(
         #: The name of the hook specification to match, see :ref:`specname`.
         self.specname = specname
 
+    def create_hookimpl(
+        self,
+        plugin: _Plugin,
+        plugin_name: str,
+        function: _HookImplFunction[object],
+    ) -> NormalImpl | WrapperImpl:
+        """Create the appropriate :class:`HookImpl` subclass for this
+        configuration.
+
+        Wrapper configurations produce a :class:`WrapperImpl`; all others
+        produce a :class:`NormalImpl`.
+        """
+        # Local import to avoid a circular import with the implementation
+        # module.
+        from ._implementation import NormalImpl
+        from ._implementation import WrapperImpl
+
+        if self.wrapper or self.hookwrapper:
+            return WrapperImpl(plugin, plugin_name, function, self)
+        return NormalImpl(plugin, plugin_name, function, self)
+
     def __repr__(self) -> str:
         attrs = [
             f"{slot}={getattr(self, slot)!r}"

src/pluggy/_execution.py

@@ -14,7 +14,6 @@
 import warnings
 
 from ._implementation import HookImpl
-from ._result import HookCallError
 from ._result import Result
 from ._warnings import PluggyTeardownRaisedWarning
 
@@ -96,12 +95,7 @@ def _multicall(
     teardowns: list[Teardown] = []
     try:  # run impl and wrapper setup functions in a loop
         for hook_impl in reversed(hook_impls):
-            try:
-                args = [caller_kwargs[argname] for argname in hook_impl.argnames]
-            except KeyError as e:
-                raise HookCallError(
-                    f"hook call must provide argument {e.args[0]!r}"
-                ) from e
+            args = hook_impl._get_call_args(caller_kwargs)
 
             if hook_impl.hookwrapper:
                 function_gen = run_old_style_hookwrapper(hook_impl, hook_name, args)

src/pluggy/_hooks.py

@@ -22,7 +22,10 @@
 from ._decorators import varnames
 from ._implementation import _HookImplFunction
 from ._implementation import _Plugin
+from ._implementation import CompletionHook
 from ._implementation import HookImpl
+from ._implementation import NormalImpl
+from ._implementation import WrapperImpl
 
 
 __all__ = [
@@ -38,6 +41,9 @@
     "_HookRelay",
     "_SubsetHookCaller",
     "HookImpl",
+    "NormalImpl",
+    "WrapperImpl",
+    "CompletionHook",
     "_HookImplFunction",
     "_Namespace",
     "_Plugin",

src/pluggy/_implementation.py

@@ -6,13 +6,18 @@
 
 from collections.abc import Callable
 from collections.abc import Generator
+from collections.abc import Mapping
+from typing import cast
 from typing import Final
 from typing import final
+from typing import Protocol
+from typing import runtime_checkable
 from typing import TypeAlias
 from typing import TypeVar
 
 from ._config import HookimplConfiguration
 from ._decorators import varnames
+from ._result import HookCallError
 from ._result import Result
 
 
@@ -22,16 +27,30 @@
 _HookImplFunction: TypeAlias = Callable[..., _T | Generator[None, Result[_T], None]]
 
 
-@final
+@runtime_checkable
+class CompletionHook(Protocol):
+    """Teardown callback returned by :meth:`WrapperImpl.setup_and_get_completion_hook`.
+
+    Receives the current ``(result, exception)`` outcome of the hook call and
+    returns the possibly replaced ``(result, exception)`` pair.
+    """
+
+    def __call__(
+        self,
+        result: object | list[object] | None,
+        exception: BaseException | None,
+    ) -> tuple[object | list[object] | None, BaseException | None]: ...
+
+
 class HookImpl:
-    """A hook implementation in a :class:`HookCaller`."""
+    """Base class for hook implementations in a :class:`HookCaller`."""
 
     __slots__ = (
         "function",
         "argnames",
         "kwargnames",
         "plugin",
-        "opts",
+        "hookimpl_config",
         "plugin_name",
         "wrapper",
         "hookwrapper",
@@ -45,7 +64,7 @@ def __init__(
         plugin: _Plugin,
         plugin_name: str,
         function: _HookImplFunction[object],
-        hook_impl_opts: HookimplConfiguration,
+        hook_impl_config: HookimplConfiguration,
     ) -> None:
         """:meta private:"""
         #: The hook implementation function.
@@ -59,23 +78,147 @@ def __init__(
         self.plugin: Final = plugin
         #: The :class:`HookimplConfiguration` used to configure this hook
         #: implementation.
-        self.opts: Final = hook_impl_opts
+        self.hookimpl_config: Final = hook_impl_config
         #: The name of the plugin which defined this hook implementation.
         self.plugin_name: Final = plugin_name
         #: Whether the hook implementation is a :ref:`wrapper <hookwrapper>`.
-        self.wrapper: Final = hook_impl_opts.wrapper
+        self.wrapper: Final = hook_impl_config.wrapper
         #: Whether the hook implementation is an :ref:`old-style wrapper
         #: <old_style_hookwrappers>`.
-        self.hookwrapper: Final = hook_impl_opts.hookwrapper
+        self.hookwrapper: Final = hook_impl_config.hookwrapper
         #: Whether validation against a hook specification is :ref:`optional
         #: <optionalhook>`.
-        self.optionalhook: Final = hook_impl_opts.optionalhook
+        self.optionalhook: Final = hook_impl_config.optionalhook
         #: Whether to try to order this hook implementation :ref:`first
         #: <callorder>`.
-        self.tryfirst: Final = hook_impl_opts.tryfirst
+        self.tryfirst: Final = hook_impl_config.tryfirst
         #: Whether to try to order this hook implementation :ref:`last
         #: <callorder>`.
-        self.trylast: Final = hook_impl_opts.trylast
+        self.trylast: Final = hook_impl_config.trylast
+
+    @property
+    def opts(self) -> HookimplConfiguration:
+        """Alias for :attr:`hookimpl_config`.
+
+        .. deprecated::
+            Use :attr:`hookimpl_config` instead.
+        """
+        return self.hookimpl_config
+
+    def _get_call_args(self, caller_kwargs: Mapping[str, object]) -> list[object]:
+        """Extract the positional arguments for calling this hook implementation.
+
+        :raises HookCallError: If a required argument is missing.
+        """
+        try:
+            return [caller_kwargs[argname] for argname in self.argnames]
+        except KeyError as e:
+            raise HookCallError(f"hook call must provide argument {e.args[0]!r}") from e
 
     def __repr__(self) -> str:
-        return f"<HookImpl plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>"
+        return (
+            f"<{type(self).__name__} "
+            f"plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>"
+        )
+
+
+@final
+class NormalImpl(HookImpl):
+    """A normal (non-wrapper) hook implementation in a :class:`HookCaller`."""
+
+    def __init__(
+        self,
+        plugin: _Plugin,
+        plugin_name: str,
+        function: _HookImplFunction[object],
+        hook_impl_config: HookimplConfiguration,
+    ) -> None:
+        """:meta private:"""
+        if hook_impl_config.wrapper or hook_impl_config.hookwrapper:
+            raise ValueError(
+                "NormalImpl cannot be used for wrapper implementations. "
+                "Use WrapperImpl instead."
+            )
+        super().__init__(plugin, plugin_name, function, hook_impl_config)
+
+
+@final
+class WrapperImpl(HookImpl):
+    """A wrapper hook implementation in a :class:`HookCaller`."""
+
+    def __init__(
+        self,
+        plugin: _Plugin,
+        plugin_name: str,
+        function: _HookImplFunction[object],
+        hook_impl_config: HookimplConfiguration,
+    ) -> None:
+        """:meta private:"""
+        if not (hook_impl_config.wrapper or hook_impl_config.hookwrapper):
+            raise ValueError(
+                "WrapperImpl can only be used for wrapper implementations. "
+                "Use NormalImpl for normal implementations."
+            )
+        super().__init__(plugin, plugin_name, function, hook_impl_config)
+
+    def setup_and_get_completion_hook(
+        self, hook_name: str, caller_kwargs: Mapping[str, object]
+    ) -> CompletionHook:
+        """Run the wrapper setup phase and return its :class:`CompletionHook`.
+
+        Old-style hookwrappers and new-style wrappers are handled uniformly by
+        adapting old-style wrappers via ``run_old_style_hookwrapper``.
+
+        The returned completion hook performs the teardown: it sends the
+        current outcome into the wrapper generator (or throws the current
+        exception) and returns the possibly replaced ``(result, exception)``
+        pair.
+        """
+        # Local import to avoid a circular import with the execution module.
+        from ._execution import _raise_wrapfail
+        from ._execution import run_old_style_hookwrapper
+
+        args = self._get_call_args(caller_kwargs)
+
+        wrapper_gen: Generator[None, object, object]
+        if self.hookwrapper:
+            wrapper_gen = run_old_style_hookwrapper(self, hook_name, args)
+        else:
+            wrapper_gen = cast(Generator[None, object, object], self.function(*args))
+
+        try:
+            next(wrapper_gen)  # first yield / setup phase
+        except StopIteration:
+            _raise_wrapfail(wrapper_gen, "did not yield")
+
+        def completion_hook(
+            result: object | list[object] | None, exception: BaseException | None
+        ) -> tuple[object | list[object] | None, BaseException | None]:
+            try:
+                if exception is not None:
+                    try:
+                        wrapper_gen.throw(exception)
+                    except RuntimeError as re:
+                        # StopIteration from generator causes RuntimeError
+                        # even for coroutine usage - see #544
+                        if (
+                            isinstance(exception, StopIteration)
+                            and re.__cause__ is exception
+                        ):
+                            wrapper_gen.close()
+                            return result, exception
+                        else:
+                            raise
+                else:
+                    wrapper_gen.send(result)
+                # Following is unreachable for a well behaved hook wrapper.
+                # Try to force finalizers otherwise postponed till GC action.
+                # Note: close() may raise if generator handles GeneratorExit.
+                wrapper_gen.close()
+                _raise_wrapfail(wrapper_gen, "has second yield")
+            except StopIteration as si:
+                return si.value, None
+            except BaseException as e:
+                return result, e
+
+        return completion_hook

src/pluggy/_manager.py

@@ -150,7 +150,7 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
             hookimpl_config = self._discover_hookimpl_configuration(plugin, name)
             if hookimpl_config is not None:
                 method: _HookImplFunction[object] = getattr(plugin, name)
-                hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_config)
+                hookimpl = hookimpl_config.create_hookimpl(plugin, plugin_name, method)
                 name = hookimpl_config.specname or name
                 hook: HookCaller | None = getattr(self.hook, name, None)
                 if hook is None:

testing/test_details.py

@@ -193,7 +193,7 @@ def myhook(self):
     plugin = Plugin()
     pname = pm.register(plugin)
     assert repr(pm.hook.myhook.get_hookimpls()[0]) == (
-        f"<HookImpl plugin_name={pname!r}, plugin={plugin!r}>"
+        f"<NormalImpl plugin_name={pname!r}, plugin={plugin!r}>"
     )