Method _select_plugin_winners

src/memos/plugins/manager.py:52–98

Resolve duplicate logical plugin names by priority. Multiple installed distributions may expose the same plugin capability (for example CE and EE variants of the Dream plugin). In that case we keep only the highest-priority implementation and skip the rest. If the h

(
        candidates: list[tuple[str, MemOSPlugin]],
    )

    @staticmethod
    def _select_plugin_winners(
        candidates: list[tuple[str, MemOSPlugin]],
    ) -> dict[str, MemOSPlugin]:
        """Resolve duplicate logical plugin names by priority.

        Multiple installed distributions may expose the same plugin capability
        (for example CE and EE variants of the Dream plugin). In that case we
        keep only the highest-priority implementation and skip the rest.

        If the highest priority is shared by more than one plugin implementation,
        startup should fail loudly because plugin activation would be ambiguous.
        """

        grouped: dict[str, list[tuple[str, MemOSPlugin]]] = {}
        for entry_point_name, plugin in candidates:
            grouped.setdefault(plugin.name, []).append((entry_point_name, plugin))

        winners: dict[str, MemOSPlugin] = {}
        for plugin_name, group in grouped.items():
            group.sort(key=lambda item: item[1].priority, reverse=True)
            winner_ep_name, winner = group[0]
            tied = [item for item in group if item[1].priority == winner.priority]
            if len(tied) > 1:
                tied_names = ", ".join(
                    f"{entry_point_name}({plugin.__class__.__name__})"
                    for entry_point_name, plugin in tied
                )
                raise RuntimeError(
                    "Multiple plugins share the same logical name and highest priority: "
                    f"name='{plugin_name}', priority={winner.priority}, providers=[{tied_names}]"
                )

            for loser_ep_name, loser in group[1:]:
                logger.info(
                    "Plugin implementation skipped due to lower priority: name=%s, "
                    "winner=%s(%s, priority=%s), skipped=%s(%s, priority=%s)",
                    plugin_name,
                    winner_ep_name,
                    winner.__class__.__name__,
                    winner.priority,
                    loser_ep_name,
                    loser.__class__.__name__,
                    loser.priority,
                )

            winners[plugin_name] = winner
        return winners

    def discover(self) -> None:
        """Discover and load all installed plugins via entry_points."""