Coverage for node / src / stigmem_node / plugins / discovery.py: 96%

1"""Plugin package discovery via Python entry points.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from collections.abc import Iterable 

7from dataclasses import dataclass 

8from importlib.metadata import EntryPoint, entry_points 

9from typing import Any 

10 

11from .errors import PluginDependencyError, PluginDiscoveryError 

12from .manifest import PluginManifest 

13 

14logger = logging.getLogger("stigmem.plugins") 

15 

16ENTRY_POINT_GROUP = "stigmem.plugins" 

17_State = str 

18_VISITING: _State = "visiting" 

19_VISITED: _State = "visited" 

20 

21# Plugins whose functionality has graduated into core. An installed copy is 

22# discovered but ignored — its hooks are never registered — so an old package 

23# cannot shadow, double-run, or fail-open against the now-core implementation. 

24# Graduated plugins add their distribution name here. 

25GRADUATED_PLUGINS: frozenset[str] = frozenset({"stigmem-plugin-memory-garden-acl"}) 

26 

27 

28@dataclass(frozen=True, slots=True) 

29class DiscoveredPlugin: 

30 """Plugin manifest loaded from a package entry point.""" 

31 

32 manifest: PluginManifest 

33 entry_point_name: str 

34 entry_point_value: str 

35 distribution: str | None = None 

36 signing_identity: str = "unsigned" 

37 signature_verified: bool = False 

38 

39 

40def discover_plugin_manifests( 

41 *, group: str = ENTRY_POINT_GROUP 

42) -> tuple[DiscoveredPlugin, ...]: 

43 """Load plugin manifests declared through Python package entry points. 

44 

45 Entry points in ``stigmem.plugins`` must resolve to a zero-argument callable 

46 returning a :class:`PluginManifest`. PR 4-INF.2 keeps discovery separate from 

47 registration so dependency ordering and startup lifecycle can build on this 

48 deterministic manifest list. 

49 """ 

50 

51 discovered: list[DiscoveredPlugin] = [] 

52 seen_names: dict[str, str] = {} 

53 for entry_point in sorted(_entry_points_for_group(group), key=lambda ep: ep.name): 

54 factory = _load_entry_point(entry_point) 

55 if not callable(factory): 

56 raise PluginDiscoveryError( 

57 f"plugin entry point {entry_point.name!r} loaded " 

58 f"{type(factory).__name__}; expected a callable returning PluginManifest" 

59 ) 

60 manifest = _call_manifest_factory(entry_point, factory) 

61 if manifest.name in GRADUATED_PLUGINS: 

62 logger.warning( 

63 "Ignoring installed plugin %r: its functionality has graduated " 

64 "into core. Uninstall the package — it is now a no-op.", 

65 manifest.name, 

66 ) 

67 continue 

68 previous_entry_point = seen_names.get(manifest.name) 

69 if previous_entry_point is not None: 

70 raise PluginDiscoveryError( 

71 f"duplicate plugin name {manifest.name!r} discovered from entry points " 

72 f"{previous_entry_point!r} and {entry_point.name!r}" 

73 ) 

74 seen_names[manifest.name] = entry_point.name 

75 discovered.append( 

76 DiscoveredPlugin( 

77 manifest=manifest, 

78 entry_point_name=entry_point.name, 

79 entry_point_value=entry_point.value, 

80 distribution=_distribution_name(entry_point), 

81 ) 

82 ) 

83 return tuple(discovered) 

84 

85 

86def resolve_plugin_dependencies( 

87 discovered: Iterable[DiscoveredPlugin], 

88 *, 

89 registered_plugins: Iterable[str] = (), 

90) -> tuple[DiscoveredPlugin, ...]: 

91 """Return discovered plugins in deterministic dependency-first order. 

92 

93 Dependencies may be satisfied by other discovered plugins or by names already 

94 present in the registry. Already-registered dependencies are not returned; 

95 they simply unlock discovered dependents for follow-on registration. 

96 """ 

97 

98 by_name: dict[str, DiscoveredPlugin] = {} 

99 for plugin in discovered: 

100 name = plugin.manifest.name 

101 if name in by_name: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 raise PluginDependencyError(f"duplicate discovered plugin name {name!r}") 

103 by_name[name] = plugin 

104 

105 registered = frozenset(registered_plugins) 

106 missing: dict[str, list[str]] = {} 

107 for name, plugin in sorted(by_name.items()): 

108 missing_deps = sorted( 

109 dependency 

110 for dependency in plugin.manifest.depends_on 

111 if dependency not in by_name and dependency not in registered 

112 ) 

113 if missing_deps: 

114 missing[name] = missing_deps 

115 if missing: 

116 details = "; ".join( 

117 f"{name} missing {', '.join(dependencies)}" 

118 for name, dependencies in missing.items() 

119 ) 

120 raise PluginDependencyError(f"missing plugin dependencies: {details}") 

121 

122 ordered: list[DiscoveredPlugin] = [] 

123 states: dict[str, _State] = {} 

124 stack: list[str] = [] 

125 

126 def visit(name: str) -> None: 

127 state = states.get(name) 

128 if state == _VISITED: 

129 return 

130 if state == _VISITING: 

131 cycle_start = stack.index(name) 

132 cycle_path = [*stack[cycle_start:], name] 

133 raise PluginDependencyError( 

134 f"plugin dependency cycle detected: {' -> '.join(cycle_path)}" 

135 ) 

136 

137 states[name] = _VISITING 

138 stack.append(name) 

139 plugin = by_name[name] 

140 for dependency in sorted(plugin.manifest.depends_on): 

141 if dependency in registered: 

142 continue 

143 visit(dependency) 

144 stack.pop() 

145 states[name] = _VISITED 

146 ordered.append(plugin) 

147 

148 for name in sorted(by_name): 

149 visit(name) 

150 

151 return tuple(ordered) 

152 

153 

154def _entry_points_for_group(group: str) -> Iterable[EntryPoint]: 

155 all_entry_points = entry_points() 

156 if hasattr(all_entry_points, "select"): 156 ↛ 158line 156 didn't jump to line 158 because the condition on line 156 was always true

157 return all_entry_points.select(group=group) 

158 return all_entry_points.get(group, ()) 

159 

160 

161def _load_entry_point(entry_point: EntryPoint) -> Any: 

162 try: 

163 return entry_point.load() 

164 except Exception as exc: 

165 raise PluginDiscoveryError( 

166 f"failed to load plugin entry point {entry_point.name!r}: {exc}" 

167 ) from exc 

168 

169 

170def _call_manifest_factory(entry_point: EntryPoint, factory: Any) -> PluginManifest: 

171 try: 

172 manifest = factory() 

173 except Exception as exc: 

174 raise PluginDiscoveryError( 

175 f"plugin entry point {entry_point.name!r} failed while creating " 

176 f"PluginManifest: {exc}" 

177 ) from exc 

178 if not isinstance(manifest, PluginManifest): 

179 raise PluginDiscoveryError( 

180 f"plugin entry point {entry_point.name!r} returned " 

181 f"{type(manifest).__name__}; expected PluginManifest" 

182 ) 

183 return manifest 

184 

185 

186def _distribution_name(entry_point: EntryPoint) -> str | None: 

187 dist = getattr(entry_point, "dist", None) 

188 if dist is None: 

189 return None 

190 metadata = getattr(dist, "metadata", None) 

191 if metadata is None: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 return None 

193 name = metadata.get("Name") 

194 return str(name) if name else None