Coverage for src/lumen/cortex/loader.py: 60%

1# TODO Adapt cortex plugin loader to use specific naming syntax 

2# TODO When loading analytics plugin it fetches duckdb as plugin environment, 

3# we should support local install by using duckdb wheel 

4""" 

5Getting plugins into the process — by configuration, never by editing the core. 

6 

7Two sources, both resolved at startup: 

8 

9* ``LUMEN_CORTEX_PLUGINS`` — a comma-separated list of ``module:ClassName`` 

10 (or just ``module``, if the module exposes exactly one ``Plugin`` subclass). 

11* Python entry points in the ``lumen.cortex.plugins`` group, so an installed 

12 package can register itself. 

13 

14A plugin that fails to load is logged and skipped. One broken extension must 

15not stop a conformant FHIR server from starting. 

16""" 

17 

18from __future__ import annotations 

19 

20import importlib 

21import inspect 

22from importlib.metadata import entry_points 

23from typing import Any 

24 

25import structlog 

26 

27from ..config import CortexSettings 

28from .contract import PluginRegistrationError 

29from .plugin import Plugin 

30from .registry import get_plugin_registry 

31 

32logger = structlog.get_logger(__name__) 

33 

34ENTRY_POINT_GROUP = "lumen.cortex.plugins" 

35 

36 

37async def load_plugins(settings: CortexSettings) -> None: 

38 """Load and register everything configuration asks for.""" 

39 if not settings.enabled: 

40 await logger.ainfo("cortex.disabled") 

41 return 

42 

43 specs = [spec.strip() for spec in settings.plugins.split(",") if spec.strip()] 

44 _ensure_plugin_requirements(specs) 

45 for spec in specs: 

46 await _load_one(spec) 

47 

48 if settings.load_entry_points: 

49 for ep in entry_points(group=ENTRY_POINT_GROUP): 

50 await _register(ep.load(), source=f"entry-point:{ep.name}") 

51 

52 registry = get_plugin_registry() 

53 await logger.ainfo( 

54 "cortex.ready", 

55 plugins=[p.name for p in registry.plugins], 

56 hooks=registry.has_hooks, 

57 operations=len(registry.operations), 

58 ) 

59 

60 

61async def _load_one(spec: str) -> None: 

62 """Load ``module:ClassName`` or ``module``.""" 

63 module_name, _, class_name = spec.partition(":") 

64 try: 

65 module = importlib.import_module(module_name) 

66 except ImportError as exc: 

67 await logger.aerror("cortex.plugin_import_failed", spec=spec, error=str(exc)) 

68 return 

69 

70 try: 

71 factory = getattr(module, class_name) if class_name else _sole_plugin_class(module, spec) 

72 except (AttributeError, PluginRegistrationError) as exc: 

73 await logger.aerror("cortex.plugin_not_found", spec=spec, error=str(exc)) 

74 return 

75 

76 await _register(factory, source=spec) 

77 

78 

79def _sole_plugin_class(module: Any, spec: str) -> type[Plugin]: 

80 candidates = [ 

81 obj 

82 for _, obj in inspect.getmembers(module, inspect.isclass) 

83 if issubclass(obj, Plugin) and obj is not Plugin and obj.__module__ == module.__name__ 

84 ] 

85 if len(candidates) != 1: 

86 raise PluginRegistrationError( 

87 f"{spec} exposes {len(candidates)} Plugin classes; name one as 'module:ClassName'" 

88 ) 

89 return candidates[0] 

90 

91 

92async def _register(factory: Any, *, source: str) -> None: 

93 """Instantiate if needed, run setup, register. Failures are logged, not raised.""" 

94 try: 

95 plugin = factory() if isinstance(factory, type) else factory 

96 if not isinstance(plugin, Plugin): 

97 raise PluginRegistrationError(f"{source} is not a lumen.cortex.Plugin") 

98 await plugin.setup() 

99 get_plugin_registry().register(plugin) 

100 except Exception as exc: 

101 await logger.aerror( 

102 "cortex.plugin_load_failed", 

103 source=source, 

104 error=str(exc), 

105 error_type=type(exc).__name__, 

106 ) 

107 

108 

109async def unload_plugins() -> None: 

110 """Tear every plugin down (server shutdown).""" 

111 registry = get_plugin_registry() 

112 for plugin in registry.plugins: 

113 try: 

114 await plugin.teardown() 

115 except Exception as exc: 

116 await logger.aerror("cortex.plugin_teardown_failed", plugin=plugin.name, error=str(exc)) 

117 registry.clear() 

118 

119 

120def _ensure_plugin_requirements(specs: list[str]) -> None: 

121 """ 

122 Check for requirements.txt in active plugin directories and auto-install 

123 (offline with wheels if available). 

124 """ 

125 import shutil 

126 import subprocess 

127 import sys 

128 from pathlib import Path 

129 

130 possible_paths: list[Path] = [ 

131 Path("/app/plugins/requirements.txt"), 

132 Path("plugins/requirements.txt"), 

133 ] 

134 

135 # Check for internal plugins if configured 

136 has_internal_plugin = any("internal_plugins" in spec for spec in specs) 

137 if has_internal_plugin: 

138 possible_paths.extend( 

139 [ 

140 Path("/app/src/lumen/cortex/internal_plugins/analytics/requirements.txt"), 

141 Path("src/lumen/cortex/internal_plugins/analytics/requirements.txt"), 

142 Path("/app/src/lumen/cortex/internal_plugins/requirements.txt"), 

143 Path("src/lumen/cortex/internal_plugins/requirements.txt"), 

144 ] 

145 ) 

146 

147 for req_file in possible_paths: 

148 if req_file.exists(): 

149 try: 

150 # Check for offline wheels directory or adjacent wheels 

151 wheel_dirs: list[Path] = [] 

152 for candidate in [req_file.parent / "wheels", req_file.parent]: 

153 if candidate.exists() and any(candidate.glob("*.whl")): 

154 wheel_dirs.append(candidate) 

155 

156 uv_bin = shutil.which("uv") 

157 if uv_bin: 

158 cmd = [ 

159 uv_bin, 

160 "pip", 

161 "install", 

162 "--cache-dir", 

163 "/tmp/uv_cache", 

164 "-q", 

165 ] 

166 else: 

167 cmd = [sys.executable, "-m", "pip", "install", "-q"] 

168 

169 if wheel_dirs: 

170 cmd.append("--no-index") 

171 for w_dir in wheel_dirs: 

172 cmd.extend(["--find-links", str(w_dir)]) 

173 

174 cmd.extend(["-r", str(req_file)]) 

175 

176 try: 

177 subprocess.check_call(cmd) 

178 except subprocess.CalledProcessError as sub_exc: 

179 if wheel_dirs: 

180 # Fallback: retry without --no-index if wheels were for other architecture 

181 structlog.get_logger(__name__).warning( 

182 "cortex.plugin_requirements_offline_failed_retrying_online", 

183 path=str(req_file), 

184 error=str(sub_exc), 

185 ) 

186 fallback_cmd = ( 

187 [ 

188 uv_bin, 

189 "pip", 

190 "install", 

191 "--cache-dir", 

192 "/tmp/uv_cache", 

193 "-q", 

194 "-r", 

195 str(req_file), 

196 ] 

197 if uv_bin 

198 else [sys.executable, "-m", "pip", "install", "-q", "-r", str(req_file)] 

199 ) 

200 subprocess.check_call(fallback_cmd) 

201 else: 

202 raise 

203 

204 structlog.get_logger(__name__).info( 

205 "cortex.plugin_requirements_installed", 

206 path=str(req_file), 

207 offline=bool(wheel_dirs), 

208 ) 

209 except Exception as exc: 

210 structlog.get_logger(__name__).error( 

211 "cortex.plugin_requirements_install_failed", path=str(req_file), error=str(exc) 

212 )