feat(async): greenlet Submitter and PluginManager.run_async

[RonnyPfannschmidt]

Written by Claude Opus 5.5 via Claude Code for the pluggy maintainers; I prompted it, it did the work, I read it.

Review for pytest-dev#710, step 7 of 7: #6 → #5 → #7 → #8 → #9 → #10 → #11
The diff here is this step alone; the description lives on the upstream PR.

[sourcery-ai]

Reviewer's Guide

Introduces a greenlet-based async Submitter and wires it through PluginManager and hook callers so that awaitable hook results can be optionally awaited via the new PluginManager.run_async API, along with tests, benchmarks, and packaging support for the async extra.

Sequence diagram for PluginManager.run_async awaiting hook results

sequenceDiagram
    actor User
    participant PluginManager
    participant NormalHookCaller
    participant _multicall
    participant HookImpl
    participant Submitter

    User->>PluginManager: run_async(lambda: hook.my_hook())
    PluginManager->>Submitter: run(sync_func)
    activate Submitter
    Submitter->>NormalHookCaller: sync_func()
    NormalHookCaller->>_multicall: _hookexec(..., async_submitter)
    _multicall->>HookImpl: function(*args)
    HookImpl-->>_multicall: awaitable_result
    _multicall->>Submitter: maybe_submit(awaitable_result)
    Submitter-->>_multicall: awaited_result
    _multicall-->>NormalHookCaller: results
    NormalHookCaller-->>Submitter: results
    Submitter-->>PluginManager: final_result
    PluginManager-->>User: awaited final_result

Loading

File-Level Changes

Change	Details	Files
Add greenlet-based async Submitter and async generator adapter to bridge sync hook execution with an event loop.	Implement Submitter to manage an active greenlet, conditionally await awaitables, and run synchronous callables in an async context Provide async_generator_to_sync helper to expose async generators as synchronous generators using Submitter Define error handling and lifecycle semantics for nested runs, missing greenlet, and generator close/throw paths	src/pluggy/_async.py testing/test_async.py
Thread Submitter through PluginManager, hook execution, and multicall so hook results can be awaited in async contexts.	Extend PluginManager.init with async_submitter parameter, store a manager-level Submitter, and expose PluginManager.run_async to run sync functions with async hook support Update _hookexec signature and traced_hookexec wrapper to accept and forward a Submitter into _multicall Propagate async_submitter into NormalHookCaller and HistoricHookCaller instances and their subset variants, using it in call/call_historic/call_extra	src/pluggy/_manager.py src/pluggy/_caller.py src/pluggy/_execution.py
Teach _multicall to optionally await awaitable hook results via Submitter while preserving "await-me-maybe" behavior outside async contexts.	Add async_submitter parameter to _multicall and detect Awaitable hook results Use Submitter.maybe_submit to await results when active or pass awaitables through unchanged otherwise Adjust benchmarks and unit tests to construct and pass a Submitter into _multicall	src/pluggy/_execution.py testing/benchmark.py testing/test_multicall.py
Add async/greenlet support to packaging and tooling configuration.	Define an 'async' optional dependency group that pulls in greenlet, and ensure testing group depends on greenlet Update pre-commit mypy configuration to include types-greenlet stubs Add placeholder changelog entry for the feature	pyproject.toml .pre-commit-config.yaml changelog/709.feature.rst uv.lock

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've found 2 issues

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="src/pluggy/_manager.py" line_range="139" />
<code_context>
     wrapper_impls: Sequence[WrapperImpl],
     caller_kwargs: Mapping[str, object],
     firstresult: bool,
+    async_submitter: Submitter,
 ) -> object | list[object]:
     """Execute a call into multiple python functions/methods and return the
</code_context>
<issue_to_address>
**issue (complexity):** Consider removing the async_submitter parameter from the public/internal hookexec functions and instead using the manager-owned submitter inside PluginManager to keep the async wiring encapsulated.

The extra `async_submitter` parameter on `_hookexec` / `traced_hookexec` is avoidable because `PluginManager` already owns `self._async_submitter`. You can keep the async feature but localize this concern inside `PluginManager` by:

* Removing `async_submitter` from the `_hookexec` / `traced_hookexec` signatures.
* Using `self._async_submitter` when delegating to `self._inner_hookexec`.

This makes hook callers and tracing oblivious to the submitter again, reducing API surface and coupling.

For example:

```python
class PluginManager:
    ...

    def _hookexec(
        self,
        hook_name: str,
        normal_impls: Sequence[NormalImpl],
        wrapper_impls: Sequence[WrapperImpl],
        caller_kwargs: Mapping[str, object],
        firstresult: bool,
    ) -> object | list[object]:
        # single source of truth
        async_submitter = self._async_submitter
        return self._inner_hookexec(
            hook_name,
            normal_impls,
            wrapper_impls,
            caller_kwargs,
            firstresult,
            async_submitter,
        )
```

And in `enable_tracing`:

```python
    def enable_tracing(
        self,
        before: _BeforeTrace | None = None,
        after: _AfterTrace | None = None,
    ) -> Callable[[], None]:
        ...

        oldcall = self._inner_hookexec

        def traced_hookexec(
            hook_name: str,
            normal_impls: Sequence[NormalImpl],
            wrapper_impls: Sequence[WrapperImpl],
            caller_kwargs: Mapping[str, object],
            firstresult: bool,
        ) -> object | list[object]:
            hook_impls: list[HookImpl] = [*normal_impls, *wrapper_impls]
            before(hook_name, hook_impls, caller_kwargs)
            outcome = Result.from_call(
                lambda: oldcall(
                    hook_name,
                    normal_impls,
                    wrapper_impls,
                    caller_kwargs,
                    firstresult,
                    self._async_submitter,  # use manager-owned submitter
                )
            )
            after(outcome, hook_name, hook_impls, caller_kwargs)
            return outcome.get_result()

        self._inner_hookexec = traced_hookexec
        ...
```

With this change:

* `_hookexec` and `traced_hookexec` keep their thinner, pre-async signatures.
* Only `PluginManager` knows about `async_submitter`; callers of `_hookexec` don’t need to pass it around.
* The async behavior is preserved because `_inner_hookexec` still receives the submitter, but the plumbing is hidden behind `PluginManager`.
</issue_to_address>

### Comment 2
<location path="src/pluggy/_caller.py" line_range="34" />
<code_context>


 _HookExec: TypeAlias = Callable[
-    [str, Sequence[NormalImpl], Sequence[WrapperImpl], Mapping[str, object], bool],
+    [
</code_context>
<issue_to_address>
**issue (complexity):** Consider keeping `_HookExec` independent of `Submitter` and binding the async submitter once in a central factory to avoid signature churn and per-caller async state.

The added `Submitter` parameter on `_HookExec` and the per-caller `_async_submitter` field do introduce avoidable coupling and signature churn. You can keep all async functionality while simplifying by binding the `Submitter` once (e.g. in `PluginManager`) and restoring the original `_HookExec` surface.

Concretely:

1. **Keep `_HookExec` unaware of `Submitter`**  
   Change `_HookExec` back to the original signature and wrap the underlying implementation with a closure/`partial` that captures the `Submitter`:

```python
# restore _HookExec type
_HookExec: TypeAlias = Callable[
    [str, Sequence[NormalImpl], Sequence[WrapperImpl], Mapping[str, object], bool],
    "object | list[object]",
]

# in PluginManager (or wherever _hookexec is created)
from functools import partial
from ._async import Submitter

def _create_hookexec(base_hookexec: Callable[..., object | list[object]],
                     submitter: Submitter) -> _HookExec:
    def hookexec(
        name: str,
        normal_impls: Sequence[NormalImpl],
        wrapper_impls: Sequence[WrapperImpl],
        kwargs: Mapping[str, object],
        firstresult: bool,
    ) -> object | list[object]:
        return base_hookexec(
            name,
            normal_impls,
            wrapper_impls,
            kwargs,
            firstresult,
            submitter,  # async plumbing stays here
        )

    return hookexec

# usage in PluginManager:
self._submitter = Submitter()
self._hookexec = _create_hookexec(base_hookexec, self._submitter)
```

2. **Drop per-caller `async_submitter` attributes and parameters**  
   Let `NormalHookCaller` and `HistoricHookCaller` just hold a bound `_hookexec` that already encapsulates the `Submitter`. Their call sites then remain simple:

```python
class NormalHookCaller:
    __slots__ = ("name", "spec", "_hookexec", "_normal_hookimpls", "_wrapper_hookimpls")

    def __init__(
        self,
        name: str,
        hook_execute: _HookExec,
        specmodule_or_class: _Namespace | None = None,
        spec_config: HookspecConfiguration | None = None,
    ) -> None:
        self.name = name
        self._hookexec = hook_execute
        # ...

    def __call__(self, **kwargs: object) -> Any:
        self._verify_all_args_are_provided(kwargs)
        firstresult = self.spec.config.firstresult if self.spec else False
        return self._hookexec(
            self.name,
            self._normal_hookimpls.copy(),
            self._wrapper_hookimpls.copy(),
            kwargs,
            firstresult,
        )
```

```python
class HistoricHookCaller:
    __slots__ = ("name", "spec", "_hookexec", "_hookimpls", "_call_history")

    def __init__(
        self,
        name: str,
        hook_execute: _HookExec,
        specmodule_or_class: _Namespace,
        spec_config: HookspecConfiguration,
    ) -> None:
        self.name = name
        self._hookexec = hook_execute
        # ...

    def call_historic(
        self,
        result_callback: Callable[[Any], None] | None = None,
        kwargs: Mapping[str, object] | None = None,
    ) -> None:
        kwargs = kwargs or {}
        self.spec.verify_all_args_are_provided(kwargs)
        self._call_history.append((kwargs, result_callback))
        res = self._hookexec(self.name, self._hookimpls.copy(), [], kwargs, False)
        # ...
```

3. **Simplify `SubsetHookCaller` to just delegate**  
   Once `_hookexec` is bound with the `Submitter`, `SubsetHookCaller` doesn’t need to reach into `orig._async_submitter` or propagate it:

```python
class SubsetHookCaller:
    def __call__(self, **kwargs: object) -> Any:
        orig = self._orig
        assert isinstance(orig, NormalHookCaller)
        if orig.spec:
            orig.spec.verify_all_args_are_provided(kwargs)
        firstresult = orig.spec.config.firstresult if orig.spec else False
        return orig._hookexec(
            self.name,
            self._get_filtered(orig._normal_hookimpls),
            self._get_filtered(orig._wrapper_hookimpls),
            kwargs,
            firstresult,
        )

    def call_historic(
        self,
        result_callback: Callable[[Any], None] | None = None,
        kwargs: Mapping[str, object] | None = None,
    ) -> None:
        orig = self._orig
        assert isinstance(orig, HistoricHookCaller)
        kwargs = kwargs or {}
        orig.spec.verify_all_args_are_provided(kwargs)
        orig._call_history.append((kwargs, result_callback))
        res = orig._hookexec(self.name, self._get_filtered(orig._hookimpls), [], kwargs, False)
        # ...
```

This keeps the async plumbing localized to the `PluginManager` (or central factory) that owns a single `Submitter`, avoids multiple instances, and restores a cleaner, decoupled hook-caller surface without changing behaviour.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue (complexity): Consider removing the async_submitter parameter from the public/internal hookexec functions and instead using the manager-owned submitter inside PluginManager to keep the async wiring encapsulated.

The extra async_submitter parameter on _hookexec / traced_hookexec is avoidable because PluginManager already owns self._async_submitter. You can keep the async feature but localize this concern inside PluginManager by:

• Removing async_submitter from the _hookexec / traced_hookexec signatures.
• Using self._async_submitter when delegating to self._inner_hookexec.

This makes hook callers and tracing oblivious to the submitter again, reducing API surface and coupling.

For example:

class PluginManager:
    ...

    def _hookexec(
        self,
        hook_name: str,
        normal_impls: Sequence[NormalImpl],
        wrapper_impls: Sequence[WrapperImpl],
        caller_kwargs: Mapping[str, object],
        firstresult: bool,
    ) -> object | list[object]:
        # single source of truth
        async_submitter = self._async_submitter
        return self._inner_hookexec(
            hook_name,
            normal_impls,
            wrapper_impls,
            caller_kwargs,
            firstresult,
            async_submitter,
        )

And in enable_tracing:

    def enable_tracing(
        self,
        before: _BeforeTrace | None = None,
        after: _AfterTrace | None = None,
    ) -> Callable[[], None]:
        ...

        oldcall = self._inner_hookexec

        def traced_hookexec(
            hook_name: str,
            normal_impls: Sequence[NormalImpl],
            wrapper_impls: Sequence[WrapperImpl],
            caller_kwargs: Mapping[str, object],
            firstresult: bool,
        ) -> object | list[object]:
            hook_impls: list[HookImpl] = [*normal_impls, *wrapper_impls]
            before(hook_name, hook_impls, caller_kwargs)
            outcome = Result.from_call(
                lambda: oldcall(
                    hook_name,
                    normal_impls,
                    wrapper_impls,
                    caller_kwargs,
                    firstresult,
                    self._async_submitter,  # use manager-owned submitter
                )
            )
            after(outcome, hook_name, hook_impls, caller_kwargs)
            return outcome.get_result()

        self._inner_hookexec = traced_hookexec
        ...

With this change:

• _hookexec and traced_hookexec keep their thinner, pre-async signatures.
• Only PluginManager knows about async_submitter; callers of _hookexec don’t need to pass it around.
• The async behavior is preserved because _inner_hookexec still receives the submitter, but the plumbing is hidden behind PluginManager.

[sourcery-ai]

issue (complexity): Consider keeping _HookExec independent of Submitter and binding the async submitter once in a central factory to avoid signature churn and per-caller async state.

The added Submitter parameter on _HookExec and the per-caller _async_submitter field do introduce avoidable coupling and signature churn. You can keep all async functionality while simplifying by binding the Submitter once (e.g. in PluginManager) and restoring the original _HookExec surface.

Concretely:

1. Keep _HookExec unaware of Submitter
   Change _HookExec back to the original signature and wrap the underlying implementation with a closure/partial that captures the Submitter:

# restore _HookExec type
_HookExec: TypeAlias = Callable[
    [str, Sequence[NormalImpl], Sequence[WrapperImpl], Mapping[str, object], bool],
    "object | list[object]",
]

# in PluginManager (or wherever _hookexec is created)
from functools import partial
from ._async import Submitter

def _create_hookexec(base_hookexec: Callable[..., object | list[object]],
                     submitter: Submitter) -> _HookExec:
    def hookexec(
        name: str,
        normal_impls: Sequence[NormalImpl],
        wrapper_impls: Sequence[WrapperImpl],
        kwargs: Mapping[str, object],
        firstresult: bool,
    ) -> object | list[object]:
        return base_hookexec(
            name,
            normal_impls,
            wrapper_impls,
            kwargs,
            firstresult,
            submitter,  # async plumbing stays here
        )

    return hookexec

# usage in PluginManager:
self._submitter = Submitter()
self._hookexec = _create_hookexec(base_hookexec, self._submitter)

2. Drop per-caller async_submitter attributes and parameters
   Let NormalHookCaller and HistoricHookCaller just hold a bound _hookexec that already encapsulates the Submitter. Their call sites then remain simple:

class NormalHookCaller:
    __slots__ = ("name", "spec", "_hookexec", "_normal_hookimpls", "_wrapper_hookimpls")

    def __init__(
        self,
        name: str,
        hook_execute: _HookExec,
        specmodule_or_class: _Namespace | None = None,
        spec_config: HookspecConfiguration | None = None,
    ) -> None:
        self.name = name
        self._hookexec = hook_execute
        # ...

    def __call__(self, **kwargs: object) -> Any:
        self._verify_all_args_are_provided(kwargs)
        firstresult = self.spec.config.firstresult if self.spec else False
        return self._hookexec(
            self.name,
            self._normal_hookimpls.copy(),
            self._wrapper_hookimpls.copy(),
            kwargs,
            firstresult,
        )

class HistoricHookCaller:
    __slots__ = ("name", "spec", "_hookexec", "_hookimpls", "_call_history")

    def __init__(
        self,
        name: str,
        hook_execute: _HookExec,
        specmodule_or_class: _Namespace,
        spec_config: HookspecConfiguration,
    ) -> None:
        self.name = name
        self._hookexec = hook_execute
        # ...

    def call_historic(
        self,
        result_callback: Callable[[Any], None] | None = None,
        kwargs: Mapping[str, object] | None = None,
    ) -> None:
        kwargs = kwargs or {}
        self.spec.verify_all_args_are_provided(kwargs)
        self._call_history.append((kwargs, result_callback))
        res = self._hookexec(self.name, self._hookimpls.copy(), [], kwargs, False)
        # ...

3. Simplify SubsetHookCaller to just delegate
   Once _hookexec is bound with the Submitter, SubsetHookCaller doesn’t need to reach into orig._async_submitter or propagate it:

class SubsetHookCaller:
    def __call__(self, **kwargs: object) -> Any:
        orig = self._orig
        assert isinstance(orig, NormalHookCaller)
        if orig.spec:
            orig.spec.verify_all_args_are_provided(kwargs)
        firstresult = orig.spec.config.firstresult if orig.spec else False
        return orig._hookexec(
            self.name,
            self._get_filtered(orig._normal_hookimpls),
            self._get_filtered(orig._wrapper_hookimpls),
            kwargs,
            firstresult,
        )

    def call_historic(
        self,
        result_callback: Callable[[Any], None] | None = None,
        kwargs: Mapping[str, object] | None = None,
    ) -> None:
        orig = self._orig
        assert isinstance(orig, HistoricHookCaller)
        kwargs = kwargs or {}
        orig.spec.verify_all_args_are_provided(kwargs)
        orig._call_history.append((kwargs, result_callback))
        res = orig._hookexec(self.name, self._get_filtered(orig._hookimpls), [], kwargs, False)
        # ...

This keeps the async plumbing localized to the PluginManager (or central factory) that owns a single Submitter, avoids multiple instances, and restores a cleaner, decoupled hook-caller surface without changing behaviour.