From 406ff6af8559d40939668050e4fb551d6b375874 Mon Sep 17 00:00:00 2001
From: starx710 <342042680+starx710@users.noreply.github.com>
Date: Tue, 13 Oct 2026 23:02:21 +0800
Subject: [PATCH 1/8] Validate hook specifications before installing them

---
 changelog/761.bugfix.rst            |   1 +
 src/pluggy/_manager.py              |  32 +++--
 testing/test_hookspec_validation.py | 183 ++++++++++++++++++++++++++++
 3 files changed, 206 insertions(+), 10 deletions(-)
 create mode 100644 changelog/761.bugfix.rst
 create mode 100644 testing/test_hookspec_validation.py

diff --git a/changelog/761.bugfix.rst b/changelog/761.bugfix.rst
new file mode 100644
index 00000000..33d1fc65
--- /dev/null
+++ b/changelog/761.bugfix.rst
@@ -0,0 +1 @@
+:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now validates all specifications before installing any of them, so failed validation no longer leaves partially installed specifications that prevent a retry. Existing hook caller references and registered implementations are preserved.
diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py
index 4994c867..308953a1 100644
--- a/src/pluggy/_manager.py
+++ b/src/pluggy/_manager.py
@@ -343,27 +343,39 @@ def add_hookspecs(self, module_or_class: _Namespace) -> None:
 
         Functions are recognized as hook specifications if they have been
         decorated with a matching :class:`HookspecMarker`.
+
+        All specifications are validated before any are installed.
         """
-        names = []
+        pending: list[tuple[str, HookCaller | None, HookCaller]] = []
         for name in dir(module_or_class):
             spec_opts = self.parse_hookspec_opts(module_or_class, name)
             if spec_opts is not None:
                 hc: HookCaller | None = getattr(self.hook, name, None)
-                if hc is None:
-                    hc = HookCaller(name, self._hookexec, module_or_class, spec_opts)
-                    setattr(self.hook, name, hc)
-                else:
-                    # Plugins registered this hook without knowing the spec.
-                    hc.set_specification(module_or_class, spec_opts)
+                if hc is not None and hc.spec is not None:
+                    raise ValueError(
+                        f"Hook {hc.spec.name!r} is already registered "
+                        f"within namespace {hc.spec.namespace}"
+                    )
+                candidate = HookCaller(name, self._hookexec, module_or_class, spec_opts)
+                if hc is not None:
+                    # Validate without installing the spec on the live caller.
                     for hookfunction in hc.get_hookimpls():
-                        self._verify_hook(hc, hookfunction)
-                names.append(name)
+                        self._verify_hook(candidate, hookfunction)
+                pending.append((name, hc, candidate))
 
-        if not names:
+        if not pending:
             raise ValueError(
                 f"did not find any {self.project_name!r} hooks in {module_or_class!r}"
             )
 
+        for name, hc, candidate in pending:
+            if hc is None:
+                setattr(self.hook, name, candidate)
+            else:
+                # Preserve references held by plugins and subset hook callers.
+                hc.spec = candidate.spec
+                hc._call_history = candidate._call_history
+
     def parse_hookspec_opts(
         self, module_or_class: _Namespace, name: str
     ) -> HookspecOpts | None:
diff --git a/testing/test_hookspec_validation.py b/testing/test_hookspec_validation.py
new file mode 100644
index 00000000..8daf3550
--- /dev/null
+++ b/testing/test_hookspec_validation.py
@@ -0,0 +1,183 @@
+"""Failed specification validation must leave registered hook callers usable."""
+
+import warnings
+
+import pytest
+
+from pluggy import HookimplMarker
+from pluggy import HookspecMarker
+from pluggy import PluginManager
+from pluggy import PluginValidationError
+
+
+hookspec = HookspecMarker("example")
+hookimpl = HookimplMarker("example")
+
+
+@pytest.mark.parametrize("historic", [False, True])
+def test_failed_hookspec_can_be_retried(pm: PluginManager, historic: bool) -> None:
+    class Hooks:
+        @hookspec(historic=historic)
+        def process(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class ValidPlugin:
+        @hookimpl
+        def process(self, value: int) -> int:
+            return value + 1
+
+    class InvalidPlugin:
+        @hookimpl
+        def process(self, unexpected: int) -> int:
+            return unexpected
+
+    valid, invalid = ValidPlugin(), InvalidPlugin()
+    pm.register(valid)
+    pm.register(invalid)
+    caller = pm.hook.process
+    impls = caller.get_hookimpls()
+    subset = pm.subset_hook_caller("process", remove_plugins=[invalid])
+
+    with pytest.raises(PluginValidationError, match="unexpected") as excinfo:
+        pm.add_hookspecs(Hooks)
+
+    assert excinfo.value.plugin is invalid
+    assert pm.hook.process is caller
+    assert not caller.has_spec()
+    assert not caller.is_historic()
+    assert caller.get_hookimpls() == impls
+    assert pm.get_plugins() == {valid, invalid}
+
+    pm.unregister(invalid)
+    pm.add_hookspecs(Hooks)
+    assert pm.hook.process is caller
+    assert caller.spec is not None
+    assert caller.spec.namespace is Hooks
+    assert caller.is_historic() is historic
+    if historic:
+        results: list[int] = []
+        subset.call_historic(results.append, {"value": 3})
+        pm.register(ValidPlugin())
+        assert results == [4, 4]
+    else:
+        assert caller(value=3) == [4]
+        assert subset(value=3) == [4]
+
+
+def test_failed_hookspec_namespace_installs_nothing(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+        @hookspec
+        def b_valid(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+        @hookspec
+        def z_invalid(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class ValidPlugin:
+        @hookimpl
+        def b_valid(self, value: int) -> int:
+            return value
+
+    class InvalidPlugin:
+        @hookimpl
+        def z_invalid(self, unexpected: int) -> int:
+            return unexpected
+
+    valid, invalid = ValidPlugin(), InvalidPlugin()
+    pm.register(valid)
+    pm.register(invalid)
+    valid_caller, invalid_caller = pm.hook.b_valid, pm.hook.z_invalid
+
+    with pytest.raises(PluginValidationError, match="unexpected"):
+        pm.add_hookspecs(Hooks)
+
+    assert not hasattr(pm.hook, "a_new")
+    assert not valid_caller.has_spec()
+    assert not invalid_caller.has_spec()
+    pm.unregister(invalid)
+    pm.add_hookspecs(Hooks)
+    assert pm.hook.b_valid is valid_caller
+    assert pm.hook.z_invalid is invalid_caller
+    assert valid_caller(value=7) == [7]
+    assert pm.hook.a_new() == []
+
+
+@pytest.mark.parametrize("historic", [False, True])
+def test_duplicate_hookspec_preserves_existing_state(
+    pm: PluginManager, historic: bool
+) -> None:
+    class ExistingHooks:
+        @hookspec(historic=historic)
+        def z_existing(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class MoreHooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+        @hookspec
+        def z_existing(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    pm.add_hookspecs(ExistingHooks)
+    caller = pm.hook.z_existing
+    original_spec = caller.spec
+    results: list[int] = []
+    if historic:
+        caller.call_historic(results.append, {"value": 8})
+
+    with pytest.raises(ValueError, match="Hook 'z_existing' is already registered"):
+        pm.add_hookspecs(MoreHooks)
+
+    assert pm.hook.z_existing is caller
+    assert caller.spec is original_spec
+    assert not hasattr(pm.hook, "a_new")
+
+    class Plugin:
+        @hookimpl
+        def z_existing(self, value: int) -> int:
+            return value
+
+    pm.register(Plugin())
+    if historic:
+        assert results == [8]
+    else:
+        assert caller(value=8) == [8]
+
+
+@pytest.mark.parametrize("warn_on_argument", [False, True])
+def test_hookspec_warning_as_error_allows_retry(
+    pm: PluginManager, warn_on_argument: bool
+) -> None:
+    warning = UserWarning("deprecated implementation")
+
+    class Hooks:
+        @hookspec(
+            warn_on_impl=None if warn_on_argument else warning,
+            warn_on_impl_args={"value": warning} if warn_on_argument else None,
+        )
+        def process(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class Plugin:
+        @hookimpl
+        def process(self, value: int) -> int:
+            return value
+
+    pm.register(Plugin())
+    caller = pm.hook.process
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        with pytest.raises(UserWarning, match="deprecated implementation"):
+            pm.add_hookspecs(Hooks)
+
+    assert not caller.has_spec()
+    with pytest.warns(UserWarning, match="deprecated implementation") as recorded:
+        pm.add_hookspecs(Hooks)
+    assert len(recorded) == 1
+    assert pm.hook.process is caller
+    assert caller(value=5) == [5]

From 24dbdad8abc838ea45096b918fde0799924aa9a5 Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Tue, 13 Oct 2026 23:02:53 +0800
Subject: [PATCH 2/8] Validate hook specifications before installing them

Validate the complete namespace before installation while retaining existing caller identities. Related to pytest-dev/pluggy#761. AI-assisted with OpenAI Codex.
---
 src/pluggy/_manager.py | 32 ++++++++++++++++++++++----------
 1 file changed, 22 insertions(+), 10 deletions(-)

diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py
index 4994c867..308953a1 100644
--- a/src/pluggy/_manager.py
+++ b/src/pluggy/_manager.py
@@ -343,27 +343,39 @@ def add_hookspecs(self, module_or_class: _Namespace) -> None:
 
         Functions are recognized as hook specifications if they have been
         decorated with a matching :class:`HookspecMarker`.
+
+        All specifications are validated before any are installed.
         """
-        names = []
+        pending: list[tuple[str, HookCaller | None, HookCaller]] = []
         for name in dir(module_or_class):
             spec_opts = self.parse_hookspec_opts(module_or_class, name)
             if spec_opts is not None:
                 hc: HookCaller | None = getattr(self.hook, name, None)
-                if hc is None:
-                    hc = HookCaller(name, self._hookexec, module_or_class, spec_opts)
-                    setattr(self.hook, name, hc)
-                else:
-                    # Plugins registered this hook without knowing the spec.
-                    hc.set_specification(module_or_class, spec_opts)
+                if hc is not None and hc.spec is not None:
+                    raise ValueError(
+                        f"Hook {hc.spec.name!r} is already registered "
+                        f"within namespace {hc.spec.namespace}"
+                    )
+                candidate = HookCaller(name, self._hookexec, module_or_class, spec_opts)
+                if hc is not None:
+                    # Validate without installing the spec on the live caller.
                     for hookfunction in hc.get_hookimpls():
-                        self._verify_hook(hc, hookfunction)
-                names.append(name)
+                        self._verify_hook(candidate, hookfunction)
+                pending.append((name, hc, candidate))
 
-        if not names:
+        if not pending:
             raise ValueError(
                 f"did not find any {self.project_name!r} hooks in {module_or_class!r}"
             )
 
+        for name, hc, candidate in pending:
+            if hc is None:
+                setattr(self.hook, name, candidate)
+            else:
+                # Preserve references held by plugins and subset hook callers.
+                hc.spec = candidate.spec
+                hc._call_history = candidate._call_history
+
     def parse_hookspec_opts(
         self, module_or_class: _Namespace, name: str
     ) -> HookspecOpts | None:

From 965f304ee340f090e4622b987c3454f42fddf68b Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Tue, 13 Oct 2026 23:04:34 +0800
Subject: [PATCH 3/8] Test failed hookspec validation and retry

Cover normal and historic retry, namespace validation, duplicate specifications, and warnings treated as errors. Seven cases fail on the original base; all pass with the fix. AI-assisted with OpenAI Codex.
---
 testing/test_hookspec_validation.py | 183 ++++++++++++++++++++++++++++
 1 file changed, 183 insertions(+)
 create mode 100644 testing/test_hookspec_validation.py

diff --git a/testing/test_hookspec_validation.py b/testing/test_hookspec_validation.py
new file mode 100644
index 00000000..8daf3550
--- /dev/null
+++ b/testing/test_hookspec_validation.py
@@ -0,0 +1,183 @@
+"""Failed specification validation must leave registered hook callers usable."""
+
+import warnings
+
+import pytest
+
+from pluggy import HookimplMarker
+from pluggy import HookspecMarker
+from pluggy import PluginManager
+from pluggy import PluginValidationError
+
+
+hookspec = HookspecMarker("example")
+hookimpl = HookimplMarker("example")
+
+
+@pytest.mark.parametrize("historic", [False, True])
+def test_failed_hookspec_can_be_retried(pm: PluginManager, historic: bool) -> None:
+    class Hooks:
+        @hookspec(historic=historic)
+        def process(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class ValidPlugin:
+        @hookimpl
+        def process(self, value: int) -> int:
+            return value + 1
+
+    class InvalidPlugin:
+        @hookimpl
+        def process(self, unexpected: int) -> int:
+            return unexpected
+
+    valid, invalid = ValidPlugin(), InvalidPlugin()
+    pm.register(valid)
+    pm.register(invalid)
+    caller = pm.hook.process
+    impls = caller.get_hookimpls()
+    subset = pm.subset_hook_caller("process", remove_plugins=[invalid])
+
+    with pytest.raises(PluginValidationError, match="unexpected") as excinfo:
+        pm.add_hookspecs(Hooks)
+
+    assert excinfo.value.plugin is invalid
+    assert pm.hook.process is caller
+    assert not caller.has_spec()
+    assert not caller.is_historic()
+    assert caller.get_hookimpls() == impls
+    assert pm.get_plugins() == {valid, invalid}
+
+    pm.unregister(invalid)
+    pm.add_hookspecs(Hooks)
+    assert pm.hook.process is caller
+    assert caller.spec is not None
+    assert caller.spec.namespace is Hooks
+    assert caller.is_historic() is historic
+    if historic:
+        results: list[int] = []
+        subset.call_historic(results.append, {"value": 3})
+        pm.register(ValidPlugin())
+        assert results == [4, 4]
+    else:
+        assert caller(value=3) == [4]
+        assert subset(value=3) == [4]
+
+
+def test_failed_hookspec_namespace_installs_nothing(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+        @hookspec
+        def b_valid(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+        @hookspec
+        def z_invalid(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class ValidPlugin:
+        @hookimpl
+        def b_valid(self, value: int) -> int:
+            return value
+
+    class InvalidPlugin:
+        @hookimpl
+        def z_invalid(self, unexpected: int) -> int:
+            return unexpected
+
+    valid, invalid = ValidPlugin(), InvalidPlugin()
+    pm.register(valid)
+    pm.register(invalid)
+    valid_caller, invalid_caller = pm.hook.b_valid, pm.hook.z_invalid
+
+    with pytest.raises(PluginValidationError, match="unexpected"):
+        pm.add_hookspecs(Hooks)
+
+    assert not hasattr(pm.hook, "a_new")
+    assert not valid_caller.has_spec()
+    assert not invalid_caller.has_spec()
+    pm.unregister(invalid)
+    pm.add_hookspecs(Hooks)
+    assert pm.hook.b_valid is valid_caller
+    assert pm.hook.z_invalid is invalid_caller
+    assert valid_caller(value=7) == [7]
+    assert pm.hook.a_new() == []
+
+
+@pytest.mark.parametrize("historic", [False, True])
+def test_duplicate_hookspec_preserves_existing_state(
+    pm: PluginManager, historic: bool
+) -> None:
+    class ExistingHooks:
+        @hookspec(historic=historic)
+        def z_existing(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class MoreHooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+        @hookspec
+        def z_existing(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    pm.add_hookspecs(ExistingHooks)
+    caller = pm.hook.z_existing
+    original_spec = caller.spec
+    results: list[int] = []
+    if historic:
+        caller.call_historic(results.append, {"value": 8})
+
+    with pytest.raises(ValueError, match="Hook 'z_existing' is already registered"):
+        pm.add_hookspecs(MoreHooks)
+
+    assert pm.hook.z_existing is caller
+    assert caller.spec is original_spec
+    assert not hasattr(pm.hook, "a_new")
+
+    class Plugin:
+        @hookimpl
+        def z_existing(self, value: int) -> int:
+            return value
+
+    pm.register(Plugin())
+    if historic:
+        assert results == [8]
+    else:
+        assert caller(value=8) == [8]
+
+
+@pytest.mark.parametrize("warn_on_argument", [False, True])
+def test_hookspec_warning_as_error_allows_retry(
+    pm: PluginManager, warn_on_argument: bool
+) -> None:
+    warning = UserWarning("deprecated implementation")
+
+    class Hooks:
+        @hookspec(
+            warn_on_impl=None if warn_on_argument else warning,
+            warn_on_impl_args={"value": warning} if warn_on_argument else None,
+        )
+        def process(self, value: int) -> int:
+            raise NotImplementedError  # pragma: no cover
+
+    class Plugin:
+        @hookimpl
+        def process(self, value: int) -> int:
+            return value
+
+    pm.register(Plugin())
+    caller = pm.hook.process
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        with pytest.raises(UserWarning, match="deprecated implementation"):
+            pm.add_hookspecs(Hooks)
+
+    assert not caller.has_spec()
+    with pytest.warns(UserWarning, match="deprecated implementation") as recorded:
+        pm.add_hookspecs(Hooks)
+    assert len(recorded) == 1
+    assert pm.hook.process is caller
+    assert caller(value=5) == [5]

From f540ce62c3165bcfc4844035e33b689fceca42d0 Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Tue, 13 Oct 2026 23:05:45 +0800
Subject: [PATCH 4/8] Document hookspec validation recovery

---
 changelog/761.bugfix.rst | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changelog/761.bugfix.rst

diff --git a/changelog/761.bugfix.rst b/changelog/761.bugfix.rst
new file mode 100644
index 00000000..33d1fc65
--- /dev/null
+++ b/changelog/761.bugfix.rst
@@ -0,0 +1 @@
+:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now validates all specifications before installing any of them, so failed validation no longer leaves partially installed specifications that prevent a retry. Existing hook caller references and registered implementations are preserved.

From 6bda1f52b24ba34916f4ced7f684b96ac96604f9 Mon Sep 17 00:00:00 2001
From: starx710 <342042680+starx710@users.noreply.github.com>
Date: Wed, 14 Oct 2026 09:12:40 +0800
Subject: [PATCH 5/8] Roll back failed historic plugin registration

Remove only implementations and unused callers created by the failed registration. Track active registration ownership to preserve nested replacement registrations, including the same plugin object. Add regression coverage and update the existing issue 761 changelog.
---
 changelog/761.bugfix.rst              |  10 +-
 src/pluggy/_manager.py                |  57 ++++--
 testing/test_pluginmanager.py         |  26 ++-
 testing/test_registration_rollback.py | 277 ++++++++++++++++++++++++++
 4 files changed, 355 insertions(+), 15 deletions(-)
 create mode 100644 testing/test_registration_rollback.py

diff --git a/changelog/761.bugfix.rst b/changelog/761.bugfix.rst
index 33d1fc65..1c4e6e8f 100644
--- a/changelog/761.bugfix.rst
+++ b/changelog/761.bugfix.rst
@@ -1 +1,9 @@
-:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now validates all specifications before installing any of them, so failed validation no longer leaves partially installed specifications that prevent a retry. Existing hook caller references and registered implementations are preserved.
+:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now
+validates all specifications before installing any of them, so failed validation
+no longer leaves partially installed specifications that prevent a retry.
+Existing hook caller references and registered implementations are preserved.
+
+Failed historic replay during :meth:`PluginManager.register()
+<pluggy.PluginManager.register>` now removes the failed registration and its
+implementations, including unused hook callers created by that registration,
+while preserving existing state and successful changes made by replay callbacks.
diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py
index 308953a1..a9449b62 100644
--- a/src/pluggy/_manager.py
+++ b/src/pluggy/_manager.py
@@ -159,6 +159,7 @@ def __init__(self, project_name: str) -> None:
         #: The project name.
         self.project_name: Final = project_name
         self._name2plugin: Final[dict[str, _Plugin]] = {}
+        self._pending_registrations: Final[dict[str, object]] = {}
         self._plugin_distinfo: Final[
             list[tuple[_Plugin, importlib.metadata.Distribution]]
         ] = []
@@ -237,19 +238,50 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
                     verified = True
                 hookimpls.append((hook_name, hookimpl, verified))
 
+        registration = object()
+        self._pending_registrations[plugin_name] = registration
         self._name2plugin[plugin_name] = plugin
-        for hook_name, hookimpl, verified in hookimpls:
-            hook = getattr(self.hook, hook_name, None)
-            if hook is None:
-                hook = HookCaller(hook_name, self._hookexec)
-                setattr(self.hook, hook_name, hook)
-            elif not verified and hook.has_spec():
-                # An earlier impl's historic replay added this spec after the
-                # first pass. Failing here leaves the plugin partially
-                # registered, but only user code during replay can get here.
-                self._verify_hook(hook, hookimpl)
-            hook._maybe_apply_history(hookimpl)
-            hook._add_hookimpl(hookimpl)
+        added: list[tuple[HookCaller, HookImpl]] = []
+        created: dict[str, HookCaller] = {}
+        try:
+            for hook_name, hookimpl, verified in hookimpls:
+                hook = getattr(self.hook, hook_name, None)
+                if hook is None:
+                    hook = HookCaller(hook_name, self._hookexec)
+                    created[hook_name] = hook
+                    setattr(self.hook, hook_name, hook)
+                elif not verified and hook.has_spec():
+                    # An earlier impl's historic replay added this spec after
+                    # the first pass, so validate against it now.
+                    self._verify_hook(hook, hookimpl)
+                hook._maybe_apply_history(hookimpl)
+                added.append((hook, hookimpl))
+                hook._add_hookimpl(hookimpl)
+        except BaseException:
+            # Replay runs user code. Remove only this registration's impls,
+            # without calling plugin equality or undoing nested registrations.
+            for hook, hookimpl in added:
+                hook._hookimpls[:] = [
+                    impl for impl in hook._hookimpls if impl is not hookimpl
+                ]
+            for hook_name, created_hook in created.items():
+                if (
+                    getattr(self.hook, hook_name, None) is created_hook
+                    and not created_hook.has_spec()
+                    and not created_hook.get_hookimpls()
+                ):
+                    delattr(self.hook, hook_name)
+            # A callback may have unregistered and re-registered even the same
+            # object under the same name. Only remove our own registry entry.
+            if (
+                self._pending_registrations.get(plugin_name) is registration
+                and self._name2plugin.get(plugin_name) is plugin
+            ):
+                del self._name2plugin[plugin_name]
+            raise
+        finally:
+            if self._pending_registrations.get(plugin_name) is registration:
+                del self._pending_registrations[plugin_name]
         return plugin_name
 
     def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None:
@@ -316,6 +348,7 @@ def unregister(
         # if self._name2plugin[name] == None registration was blocked: ignore
         if self._name2plugin.get(name) is not None:
             del self._name2plugin[name]
+            self._pending_registrations.pop(name, None)
 
         return plugin
 
diff --git a/testing/test_pluginmanager.py b/testing/test_pluginmanager.py
index 06d645d7..10d0aa56 100644
--- a/testing/test_pluginmanager.py
+++ b/testing/test_pluginmanager.py
@@ -424,8 +424,30 @@ def configure(self):
         def late(self, qlwkje):
             pass  # pragma: no cover
 
-    with pytest.raises(PluginValidationError, match="qlwkje"):
-        pm.register(hello())
+    plugin = hello()
+    caller = pm.hook.configure
+    spec = caller.spec
+    history = caller._call_history
+    with pytest.raises(PluginValidationError, match="qlwkje") as excinfo:
+        pm.register(plugin)
+
+    assert excinfo.value.plugin is plugin
+    assert pm.list_name_plugin() == []
+    assert pm.hook.configure is caller
+    assert caller.spec is spec
+    assert caller._call_history is history
+    assert caller.get_hookimpls() == []
+    assert pm.hook.late.spec is not None
+    assert pm.hook.late.spec.namespace is LateSpecs
+    assert pm.hook.late.get_hookimpls() == []
+
+    class Valid:
+        @hookimpl
+        def late(self, arg):
+            return arg
+
+    pm.register(Valid())
+    assert pm.hook.late(arg=7) == [7]
 
 
 def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) -> None:
diff --git a/testing/test_registration_rollback.py b/testing/test_registration_rollback.py
new file mode 100644
index 00000000..93a980d2
--- /dev/null
+++ b/testing/test_registration_rollback.py
@@ -0,0 +1,277 @@
+"""Historic replay failures must only undo the failed registration."""
+
+import pytest
+
+from pluggy import HookimplMarker
+from pluggy import HookspecMarker
+from pluggy import PluginManager
+
+
+hookspec = HookspecMarker("example")
+hookimpl = HookimplMarker("example")
+
+
+@pytest.mark.parametrize("exception_type", [RuntimeError, KeyboardInterrupt])
+@pytest.mark.parametrize("fail_in_callback", [False, True])
+def test_failed_replay_cleans_up_and_allows_retry(
+    pm: PluginManager, exception_type: type[BaseException], fail_in_callback: bool
+) -> None:
+    class Hooks:
+        @hookspec
+        def a_existing(self) -> None: ...
+
+        @hookspec(historic=True)
+        def c_history(self, value: int) -> None: ...
+
+    class Existing:
+        @hookimpl
+        def a_existing(self) -> str:
+            return "existing"
+
+        @hookimpl
+        def c_history(self, value: int) -> tuple[str, int]:
+            return "existing", value
+
+    class Empty:
+        @hookimpl
+        def d_empty(self) -> None: ...
+
+    error = exception_type("replay failed")
+    fail = True
+    calls: list[int] = []
+    results: list[tuple[str, int]] = []
+
+    def callback(result: tuple[str, int]) -> None:
+        results.append(result)
+        if fail and fail_in_callback and result == ("new", 20):
+            raise error
+
+    class Plugin:
+        @hookimpl
+        def a_existing(self) -> str:
+            return "new"
+
+        @hookimpl
+        def b_new(self) -> str:
+            return "new hook"
+
+        @hookimpl
+        def c_history(self, value: int) -> tuple[str, int]:
+            calls.append(value)
+            if fail and not fail_in_callback and value == 20:
+                raise error
+            return "new", value
+
+        @hookimpl
+        def z_unprocessed(self) -> None: ...
+
+    pm.add_hookspecs(Hooks)
+    existing = Existing()
+    pm.register(existing, "existing")
+    empty = Empty()
+    pm.register(empty, "empty")
+    pm.unregister(empty)
+    callers = dict(pm.hook.__dict__)
+    impls = {name: caller.get_hookimpls() for name, caller in callers.items()}
+    specs = {name: caller.spec for name, caller in callers.items()}
+    history = pm.hook.c_history._call_history
+    for value in (10, 20):
+        pm.hook.c_history.call_historic(callback, {"value": value})
+    plugin = Plugin()
+
+    with pytest.raises(exception_type, match="replay failed") as excinfo:
+        pm.register(plugin, "new")
+
+    assert excinfo.value is error
+    frame_names = [entry.name for entry in excinfo.traceback]
+    assert frame_names[-1] == ("callback" if fail_in_callback else "c_history")
+    assert frame_names.count("register") == 1
+    assert pm.list_name_plugin() == [("existing", existing)]
+    assert pm.get_hookcallers(plugin) is None
+    assert pm.hook.__dict__ == callers
+    for name, caller in callers.items():
+        assert getattr(pm.hook, name) is caller
+        assert caller.spec is specs[name]
+        assert caller.get_hookimpls() == impls[name]
+    assert pm.hook.c_history._call_history is history
+    assert history is not None and len(history) == 2
+    assert calls == [10, 20]  # External effects of replay are retained.
+    assert results == [("existing", 10), ("existing", 20), ("new", 10)] + (
+        [("new", 20)] if fail_in_callback else []
+    )
+    assert pm.hook.a_existing() == ["existing"]
+
+    fail = False
+    assert pm.register(plugin, "new") == "new"
+    assert calls == [10, 20, 10, 20]
+    assert results[-2:] == [("new", 10), ("new", 20)]
+    assert pm.hook.a_existing() == ["new", "existing"]
+    assert pm.hook.b_new() == ["new hook"]
+    assert pm.hook.z_unprocessed() == []
+    pm.unregister(plugin)
+    assert pm.list_name_plugin() == [("existing", existing)]
+    for name, caller in callers.items():
+        assert caller.get_hookimpls() == impls[name]
+
+
+@pytest.mark.parametrize("adopted_by", ["spec", "plugin", "both"])
+def test_failed_replay_preserves_adopted_new_caller(
+    pm: PluginManager, adopted_by: str
+) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    class LateHooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+    class Nested:
+        @hookimpl
+        def a_new(self) -> str:
+            return "nested"
+
+    nested = Nested()
+    error = RuntimeError("outer replay failed")
+    saved = []
+
+    class Plugin:
+        @hookimpl
+        def a_new(self) -> str:
+            return "outer"  # pragma: no cover
+
+        @hookimpl
+        def configure(self) -> None:
+            caller = pm.hook.a_new
+            if adopted_by in ("spec", "both"):
+                pm.add_hookspecs(LateHooks)
+            if adopted_by in ("plugin", "both"):
+                pm.register(nested, "nested")
+            saved.append((caller, caller.spec, caller.get_hookimpls()[-1:]))
+            raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(Plugin(), "outer")
+
+    assert excinfo.value is error
+    caller, spec, nested_impls = saved[0]
+    assert pm.hook.a_new is caller
+    assert caller.spec is spec
+    assert caller.has_spec() is (adopted_by != "plugin")
+    assert pm.hook.configure.get_hookimpls() == []
+    if adopted_by == "spec":
+        assert pm.list_name_plugin() == []
+        assert caller.get_hookimpls() == []
+        assert caller() == []
+    else:
+        assert pm.list_name_plugin() == [("nested", nested)]
+        assert caller.get_hookimpls() == nested_impls
+        assert caller() == ["nested"]
+
+
+@pytest.mark.parametrize(
+    "replacement", ["other", "same", "same_without_hooks", "blocked", "removed"]
+)
+def test_failed_replay_preserves_reentrant_registry_changes(
+    pm: PluginManager, replacement: str
+) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    class Other:
+        @hookimpl
+        def a_new(self) -> str:
+            return "other"
+
+    other = Other()
+    error = RuntimeError("replay replaced registration")
+
+    class Plugin:
+        replace = True
+
+        @hookimpl
+        def a_new(self) -> str:
+            return "same"
+
+        @hookimpl
+        def configure(self) -> None:
+            if self.replace:
+                self.replace = False
+                pm.unregister(name="plugin")
+                if replacement == "other":
+                    pm.register(other, "plugin")
+                elif replacement == "same":
+                    pm.register(self, "plugin")
+                elif replacement == "same_without_hooks":
+                    self.a_new = None  # type: ignore[assignment]
+                    self.configure = None  # type: ignore[assignment]
+                    pm.register(self, "plugin")
+                elif replacement == "blocked":
+                    pm.set_blocked("plugin")
+                raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    plugin = Plugin()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(plugin, "plugin")
+
+    assert excinfo.value is error
+    if replacement == "other":
+        assert pm.list_name_plugin() == [("plugin", other)]
+        assert pm.hook.a_new() == ["other"]
+        assert pm.hook.configure.get_hookimpls() == []
+    elif replacement == "same":
+        assert pm.list_name_plugin() == [("plugin", plugin)]
+        assert pm.hook.a_new() == ["same"]
+        assert len(pm.hook.configure.get_hookimpls()) == 1
+        pm.unregister(plugin)
+        assert pm.hook.a_new.get_hookimpls() == []
+        assert pm.hook.configure.get_hookimpls() == []
+    elif replacement == "same_without_hooks":
+        assert pm.list_name_plugin() == [("plugin", plugin)]
+        assert pm.get_hookcallers(plugin) == []
+        assert not hasattr(pm.hook, "a_new")
+        assert pm.hook.configure.get_hookimpls() == []
+    else:
+        assert pm.get_plugin("plugin") is None
+        assert pm.is_blocked("plugin") is (replacement == "blocked")
+        assert not hasattr(pm.hook, "a_new")
+        assert pm.hook.configure.get_hookimpls() == []
+
+
+def test_rollback_does_not_call_plugin_equality(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    error = RuntimeError("original replay failure")
+
+    class Plugin:
+        fail = False
+
+        def __eq__(self, other: object) -> bool:
+            if self.fail:
+                raise AssertionError("equality must not mask the replay error")
+            return self is other
+
+        @hookimpl
+        def a_new(self) -> None: ...
+
+        @hookimpl
+        def configure(self) -> None:
+            self.fail = True
+            raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(Plugin(), "plugin")
+
+    assert excinfo.value is error
+    assert pm.list_name_plugin() == []
+    assert not hasattr(pm.hook, "a_new")
+    assert pm.hook.configure.get_hookimpls() == []

From f1855d06397bef029d89a11a4fbde934acc7609b Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Wed, 14 Oct 2026 09:14:59 +0800
Subject: [PATCH 6/8] Roll back failed historic plugin registration

Remove only implementations and unused callers created by the failed registration. Preserve nested registry changes, including re-registration of the same plugin object.
---
 src/pluggy/_manager.py | 57 +++++++++++++++++++++++++++++++++---------
 1 file changed, 45 insertions(+), 12 deletions(-)

diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py
index 308953a1..a9449b62 100644
--- a/src/pluggy/_manager.py
+++ b/src/pluggy/_manager.py
@@ -159,6 +159,7 @@ def __init__(self, project_name: str) -> None:
         #: The project name.
         self.project_name: Final = project_name
         self._name2plugin: Final[dict[str, _Plugin]] = {}
+        self._pending_registrations: Final[dict[str, object]] = {}
         self._plugin_distinfo: Final[
             list[tuple[_Plugin, importlib.metadata.Distribution]]
         ] = []
@@ -237,19 +238,50 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
                     verified = True
                 hookimpls.append((hook_name, hookimpl, verified))
 
+        registration = object()
+        self._pending_registrations[plugin_name] = registration
         self._name2plugin[plugin_name] = plugin
-        for hook_name, hookimpl, verified in hookimpls:
-            hook = getattr(self.hook, hook_name, None)
-            if hook is None:
-                hook = HookCaller(hook_name, self._hookexec)
-                setattr(self.hook, hook_name, hook)
-            elif not verified and hook.has_spec():
-                # An earlier impl's historic replay added this spec after the
-                # first pass. Failing here leaves the plugin partially
-                # registered, but only user code during replay can get here.
-                self._verify_hook(hook, hookimpl)
-            hook._maybe_apply_history(hookimpl)
-            hook._add_hookimpl(hookimpl)
+        added: list[tuple[HookCaller, HookImpl]] = []
+        created: dict[str, HookCaller] = {}
+        try:
+            for hook_name, hookimpl, verified in hookimpls:
+                hook = getattr(self.hook, hook_name, None)
+                if hook is None:
+                    hook = HookCaller(hook_name, self._hookexec)
+                    created[hook_name] = hook
+                    setattr(self.hook, hook_name, hook)
+                elif not verified and hook.has_spec():
+                    # An earlier impl's historic replay added this spec after
+                    # the first pass, so validate against it now.
+                    self._verify_hook(hook, hookimpl)
+                hook._maybe_apply_history(hookimpl)
+                added.append((hook, hookimpl))
+                hook._add_hookimpl(hookimpl)
+        except BaseException:
+            # Replay runs user code. Remove only this registration's impls,
+            # without calling plugin equality or undoing nested registrations.
+            for hook, hookimpl in added:
+                hook._hookimpls[:] = [
+                    impl for impl in hook._hookimpls if impl is not hookimpl
+                ]
+            for hook_name, created_hook in created.items():
+                if (
+                    getattr(self.hook, hook_name, None) is created_hook
+                    and not created_hook.has_spec()
+                    and not created_hook.get_hookimpls()
+                ):
+                    delattr(self.hook, hook_name)
+            # A callback may have unregistered and re-registered even the same
+            # object under the same name. Only remove our own registry entry.
+            if (
+                self._pending_registrations.get(plugin_name) is registration
+                and self._name2plugin.get(plugin_name) is plugin
+            ):
+                del self._name2plugin[plugin_name]
+            raise
+        finally:
+            if self._pending_registrations.get(plugin_name) is registration:
+                del self._pending_registrations[plugin_name]
         return plugin_name
 
     def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None:
@@ -316,6 +348,7 @@ def unregister(
         # if self._name2plugin[name] == None registration was blocked: ignore
         if self._name2plugin.get(name) is not None:
             del self._name2plugin[name]
+            self._pending_registrations.pop(name, None)
 
         return plugin
 

From c5e82913e13dc1042eaa6ca94bbeb9d1a6c7c5d4 Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Wed, 14 Oct 2026 09:16:10 +0800
Subject: [PATCH 7/8] Test historic registration rollback and reentrant
 recovery

Cover replay and callback failures, original tracebacks, retries, existing state, unused new callers, nested registrations and late-spec validation. Full suite: 239 passed; all pre-commit hooks passed.
---
 testing/test_pluginmanager.py         |  26 ++-
 testing/test_registration_rollback.py | 277 ++++++++++++++++++++++++++
 2 files changed, 301 insertions(+), 2 deletions(-)
 create mode 100644 testing/test_registration_rollback.py

diff --git a/testing/test_pluginmanager.py b/testing/test_pluginmanager.py
index 06d645d7..10d0aa56 100644
--- a/testing/test_pluginmanager.py
+++ b/testing/test_pluginmanager.py
@@ -424,8 +424,30 @@ def configure(self):
         def late(self, qlwkje):
             pass  # pragma: no cover
 
-    with pytest.raises(PluginValidationError, match="qlwkje"):
-        pm.register(hello())
+    plugin = hello()
+    caller = pm.hook.configure
+    spec = caller.spec
+    history = caller._call_history
+    with pytest.raises(PluginValidationError, match="qlwkje") as excinfo:
+        pm.register(plugin)
+
+    assert excinfo.value.plugin is plugin
+    assert pm.list_name_plugin() == []
+    assert pm.hook.configure is caller
+    assert caller.spec is spec
+    assert caller._call_history is history
+    assert caller.get_hookimpls() == []
+    assert pm.hook.late.spec is not None
+    assert pm.hook.late.spec.namespace is LateSpecs
+    assert pm.hook.late.get_hookimpls() == []
+
+    class Valid:
+        @hookimpl
+        def late(self, arg):
+            return arg
+
+    pm.register(Valid())
+    assert pm.hook.late(arg=7) == [7]
 
 
 def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) -> None:
diff --git a/testing/test_registration_rollback.py b/testing/test_registration_rollback.py
new file mode 100644
index 00000000..93a980d2
--- /dev/null
+++ b/testing/test_registration_rollback.py
@@ -0,0 +1,277 @@
+"""Historic replay failures must only undo the failed registration."""
+
+import pytest
+
+from pluggy import HookimplMarker
+from pluggy import HookspecMarker
+from pluggy import PluginManager
+
+
+hookspec = HookspecMarker("example")
+hookimpl = HookimplMarker("example")
+
+
+@pytest.mark.parametrize("exception_type", [RuntimeError, KeyboardInterrupt])
+@pytest.mark.parametrize("fail_in_callback", [False, True])
+def test_failed_replay_cleans_up_and_allows_retry(
+    pm: PluginManager, exception_type: type[BaseException], fail_in_callback: bool
+) -> None:
+    class Hooks:
+        @hookspec
+        def a_existing(self) -> None: ...
+
+        @hookspec(historic=True)
+        def c_history(self, value: int) -> None: ...
+
+    class Existing:
+        @hookimpl
+        def a_existing(self) -> str:
+            return "existing"
+
+        @hookimpl
+        def c_history(self, value: int) -> tuple[str, int]:
+            return "existing", value
+
+    class Empty:
+        @hookimpl
+        def d_empty(self) -> None: ...
+
+    error = exception_type("replay failed")
+    fail = True
+    calls: list[int] = []
+    results: list[tuple[str, int]] = []
+
+    def callback(result: tuple[str, int]) -> None:
+        results.append(result)
+        if fail and fail_in_callback and result == ("new", 20):
+            raise error
+
+    class Plugin:
+        @hookimpl
+        def a_existing(self) -> str:
+            return "new"
+
+        @hookimpl
+        def b_new(self) -> str:
+            return "new hook"
+
+        @hookimpl
+        def c_history(self, value: int) -> tuple[str, int]:
+            calls.append(value)
+            if fail and not fail_in_callback and value == 20:
+                raise error
+            return "new", value
+
+        @hookimpl
+        def z_unprocessed(self) -> None: ...
+
+    pm.add_hookspecs(Hooks)
+    existing = Existing()
+    pm.register(existing, "existing")
+    empty = Empty()
+    pm.register(empty, "empty")
+    pm.unregister(empty)
+    callers = dict(pm.hook.__dict__)
+    impls = {name: caller.get_hookimpls() for name, caller in callers.items()}
+    specs = {name: caller.spec for name, caller in callers.items()}
+    history = pm.hook.c_history._call_history
+    for value in (10, 20):
+        pm.hook.c_history.call_historic(callback, {"value": value})
+    plugin = Plugin()
+
+    with pytest.raises(exception_type, match="replay failed") as excinfo:
+        pm.register(plugin, "new")
+
+    assert excinfo.value is error
+    frame_names = [entry.name for entry in excinfo.traceback]
+    assert frame_names[-1] == ("callback" if fail_in_callback else "c_history")
+    assert frame_names.count("register") == 1
+    assert pm.list_name_plugin() == [("existing", existing)]
+    assert pm.get_hookcallers(plugin) is None
+    assert pm.hook.__dict__ == callers
+    for name, caller in callers.items():
+        assert getattr(pm.hook, name) is caller
+        assert caller.spec is specs[name]
+        assert caller.get_hookimpls() == impls[name]
+    assert pm.hook.c_history._call_history is history
+    assert history is not None and len(history) == 2
+    assert calls == [10, 20]  # External effects of replay are retained.
+    assert results == [("existing", 10), ("existing", 20), ("new", 10)] + (
+        [("new", 20)] if fail_in_callback else []
+    )
+    assert pm.hook.a_existing() == ["existing"]
+
+    fail = False
+    assert pm.register(plugin, "new") == "new"
+    assert calls == [10, 20, 10, 20]
+    assert results[-2:] == [("new", 10), ("new", 20)]
+    assert pm.hook.a_existing() == ["new", "existing"]
+    assert pm.hook.b_new() == ["new hook"]
+    assert pm.hook.z_unprocessed() == []
+    pm.unregister(plugin)
+    assert pm.list_name_plugin() == [("existing", existing)]
+    for name, caller in callers.items():
+        assert caller.get_hookimpls() == impls[name]
+
+
+@pytest.mark.parametrize("adopted_by", ["spec", "plugin", "both"])
+def test_failed_replay_preserves_adopted_new_caller(
+    pm: PluginManager, adopted_by: str
+) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    class LateHooks:
+        @hookspec
+        def a_new(self) -> None: ...
+
+    class Nested:
+        @hookimpl
+        def a_new(self) -> str:
+            return "nested"
+
+    nested = Nested()
+    error = RuntimeError("outer replay failed")
+    saved = []
+
+    class Plugin:
+        @hookimpl
+        def a_new(self) -> str:
+            return "outer"  # pragma: no cover
+
+        @hookimpl
+        def configure(self) -> None:
+            caller = pm.hook.a_new
+            if adopted_by in ("spec", "both"):
+                pm.add_hookspecs(LateHooks)
+            if adopted_by in ("plugin", "both"):
+                pm.register(nested, "nested")
+            saved.append((caller, caller.spec, caller.get_hookimpls()[-1:]))
+            raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(Plugin(), "outer")
+
+    assert excinfo.value is error
+    caller, spec, nested_impls = saved[0]
+    assert pm.hook.a_new is caller
+    assert caller.spec is spec
+    assert caller.has_spec() is (adopted_by != "plugin")
+    assert pm.hook.configure.get_hookimpls() == []
+    if adopted_by == "spec":
+        assert pm.list_name_plugin() == []
+        assert caller.get_hookimpls() == []
+        assert caller() == []
+    else:
+        assert pm.list_name_plugin() == [("nested", nested)]
+        assert caller.get_hookimpls() == nested_impls
+        assert caller() == ["nested"]
+
+
+@pytest.mark.parametrize(
+    "replacement", ["other", "same", "same_without_hooks", "blocked", "removed"]
+)
+def test_failed_replay_preserves_reentrant_registry_changes(
+    pm: PluginManager, replacement: str
+) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    class Other:
+        @hookimpl
+        def a_new(self) -> str:
+            return "other"
+
+    other = Other()
+    error = RuntimeError("replay replaced registration")
+
+    class Plugin:
+        replace = True
+
+        @hookimpl
+        def a_new(self) -> str:
+            return "same"
+
+        @hookimpl
+        def configure(self) -> None:
+            if self.replace:
+                self.replace = False
+                pm.unregister(name="plugin")
+                if replacement == "other":
+                    pm.register(other, "plugin")
+                elif replacement == "same":
+                    pm.register(self, "plugin")
+                elif replacement == "same_without_hooks":
+                    self.a_new = None  # type: ignore[assignment]
+                    self.configure = None  # type: ignore[assignment]
+                    pm.register(self, "plugin")
+                elif replacement == "blocked":
+                    pm.set_blocked("plugin")
+                raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    plugin = Plugin()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(plugin, "plugin")
+
+    assert excinfo.value is error
+    if replacement == "other":
+        assert pm.list_name_plugin() == [("plugin", other)]
+        assert pm.hook.a_new() == ["other"]
+        assert pm.hook.configure.get_hookimpls() == []
+    elif replacement == "same":
+        assert pm.list_name_plugin() == [("plugin", plugin)]
+        assert pm.hook.a_new() == ["same"]
+        assert len(pm.hook.configure.get_hookimpls()) == 1
+        pm.unregister(plugin)
+        assert pm.hook.a_new.get_hookimpls() == []
+        assert pm.hook.configure.get_hookimpls() == []
+    elif replacement == "same_without_hooks":
+        assert pm.list_name_plugin() == [("plugin", plugin)]
+        assert pm.get_hookcallers(plugin) == []
+        assert not hasattr(pm.hook, "a_new")
+        assert pm.hook.configure.get_hookimpls() == []
+    else:
+        assert pm.get_plugin("plugin") is None
+        assert pm.is_blocked("plugin") is (replacement == "blocked")
+        assert not hasattr(pm.hook, "a_new")
+        assert pm.hook.configure.get_hookimpls() == []
+
+
+def test_rollback_does_not_call_plugin_equality(pm: PluginManager) -> None:
+    class Hooks:
+        @hookspec(historic=True)
+        def configure(self) -> None: ...
+
+    error = RuntimeError("original replay failure")
+
+    class Plugin:
+        fail = False
+
+        def __eq__(self, other: object) -> bool:
+            if self.fail:
+                raise AssertionError("equality must not mask the replay error")
+            return self is other
+
+        @hookimpl
+        def a_new(self) -> None: ...
+
+        @hookimpl
+        def configure(self) -> None:
+            self.fail = True
+            raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.configure.call_historic()
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(Plugin(), "plugin")
+
+    assert excinfo.value is error
+    assert pm.list_name_plugin() == []
+    assert not hasattr(pm.hook, "a_new")
+    assert pm.hook.configure.get_hookimpls() == []

From 1c62d196d2111dc7d36b638bbb551b507e8aa4a9 Mon Sep 17 00:00:00 2001
From: starx710 <star710.x@gmail.com>
Date: Wed, 14 Oct 2026 09:16:55 +0800
Subject: [PATCH 8/8] Document both issue 761 failure recovery fixes

Update the existing changelog fragment for hookspec validation and historic registration rollback.
---
 changelog/761.bugfix.rst | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/changelog/761.bugfix.rst b/changelog/761.bugfix.rst
index 33d1fc65..1c4e6e8f 100644
--- a/changelog/761.bugfix.rst
+++ b/changelog/761.bugfix.rst
@@ -1 +1,9 @@
-:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now validates all specifications before installing any of them, so failed validation no longer leaves partially installed specifications that prevent a retry. Existing hook caller references and registered implementations are preserved.
+:meth:`PluginManager.add_hookspecs() <pluggy.PluginManager.add_hookspecs>` now
+validates all specifications before installing any of them, so failed validation
+no longer leaves partially installed specifications that prevent a retry.
+Existing hook caller references and registered implementations are preserved.
+
+Failed historic replay during :meth:`PluginManager.register()
+<pluggy.PluginManager.register>` now removes the failed registration and its
+implementations, including unused hook callers created by that registration,
+while preserving existing state and successful changes made by replay callbacks.