Coverage for little_loops / extension.py: 33%

1"""Extension system for little-loops. 

2 

3Provides the extension Protocol, a reference implementation, and discovery/loading 

4utilities for external packages to hook into little-loops via structured events. 

5 

6Public exports: 

7 LLExtension: Protocol that extensions must satisfy 

8 NoopLoggerExtension: Reference extension that logs events to a JSONL file 

9 ExtensionLoader: Discovers and loads extensions from config and entry points 

10""" 

11 

12from __future__ import annotations 

13 

14import importlib 

15import json 

16import logging 

17from importlib.metadata import entry_points 

18from pathlib import Path 

19from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

20 

21from little_loops.events import EventBus, EventCallback, LLEvent 

22from little_loops.issue_parser import IssueInfo 

23 

24if TYPE_CHECKING: 

25 from little_loops.fsm.executor import FSMExecutor, RouteContext, RouteDecision 

26 from little_loops.fsm.persistence import PersistentExecutor 

27 from little_loops.fsm.runners import ActionRunner 

28 from little_loops.fsm.types import Evaluator 

29 

30logger = logging.getLogger(__name__) 

31 

32ENTRY_POINT_GROUP = "little_loops.extensions" 

33 

34 

35@runtime_checkable 

36class LLExtension(Protocol): 

37 """Protocol for little-loops extensions. 

38 

39 Any class with an ``on_event`` method matching this signature is a valid 

40 extension. Extensions receive structured events from the EventBus. 

41 

42 Optionally, an extension may declare ``event_filter`` to subscribe only to 

43 specific event namespaces via glob patterns (e.g. ``"issue.*"`` or 

44 ``["issue.*", "parallel.*"]``). ``None`` (the default) means the extension 

45 receives every event. 

46 """ 

47 

48 event_filter: str | list[str] | None 

49 

50 def on_event(self, event: LLEvent) -> None: 

51 """Handle an event from little-loops. 

52 

53 Args: 

54 event: Structured event with type, timestamp, and payload 

55 """ 

56 ... 

57 

58 

59class InterceptorExtension(Protocol): 

60 """Protocol for extensions that intercept FSM routing decisions. 

61 

62 Detected via hasattr() in wire_extensions() — no @runtime_checkable needed. 

63 All methods are optional to implement individually; detection is per-method. 

64 """ 

65 

66 def before_route(self, context: RouteContext) -> RouteDecision | None: 

67 """Called before routing; return RouteDecision to redirect or veto, None to pass through.""" 

68 ... 

69 

70 def after_route(self, context: RouteContext) -> None: 

71 """Called after routing is resolved (observational only).""" 

72 ... 

73 

74 def before_issue_close(self, info: IssueInfo) -> bool | None: 

75 """Called before an issue is closed; return False to veto, None to pass through.""" 

76 ... 

77 

78 

79class ActionProviderExtension(Protocol): 

80 """Protocol for extensions that contribute custom actions to FSM loops. 

81 

82 Detected via hasattr() in wire_extensions() — no @runtime_checkable needed. 

83 """ 

84 

85 def provided_actions(self) -> dict[str, ActionRunner]: 

86 """Return a mapping of action name → ActionRunner for injection into the executor.""" 

87 ... 

88 

89 

90class EvaluatorProviderExtension(Protocol): 

91 """Protocol for extensions that contribute custom evaluators to FSM loops. 

92 

93 Detected via hasattr() in wire_extensions() — no @runtime_checkable needed. 

94 """ 

95 

96 def provided_evaluators(self) -> dict[str, Evaluator]: 

97 """Return a mapping of evaluator type name → Evaluator callable.""" 

98 ... 

99 

100 

101class NoopLoggerExtension: 

102 """Reference extension that logs events to a JSONL file. 

103 

104 Demonstrates the extension API by writing each event as a JSON line 

105 to a specified log file. Useful for debugging and as a template for 

106 building custom extensions. 

107 """ 

108 

109 def __init__(self, log_path: Path | None = None) -> None: 

110 self._log_path = log_path or Path(".ll/extension-events.jsonl") 

111 self._log_path.parent.mkdir(parents=True, exist_ok=True) 

112 

113 def on_event(self, event: LLEvent) -> None: 

114 """Append event to JSONL log file.""" 

115 with open(self._log_path, "a", encoding="utf-8") as f: 

116 f.write(json.dumps(event.to_dict()) + "\n") 

117 

118 

119class ExtensionLoader: 

120 """Discover and load extensions from config paths and entry points.""" 

121 

122 @staticmethod 

123 def from_config(extension_paths: list[str]) -> list[LLExtension]: 

124 """Load extensions from dotted module paths. 

125 

126 Each path should be in the format "module.path:ClassName". 

127 Extensions that fail to load are skipped with a warning. 

128 

129 Args: 

130 extension_paths: List of "module:Class" strings 

131 

132 Returns: 

133 List of successfully loaded extension instances 

134 """ 

135 extensions: list[Any] = [] 

136 for path in extension_paths: 

137 try: 

138 module_path, class_name = path.rsplit(":", 1) 

139 module = importlib.import_module(module_path) 

140 cls = getattr(module, class_name) 

141 extensions.append(cls()) 

142 except Exception: 

143 logger.warning("Failed to load extension from config: %s", path, exc_info=True) 

144 return extensions 

145 

146 @staticmethod 

147 def from_entry_points() -> list[LLExtension]: 

148 """Discover extensions via importlib.metadata entry points. 

149 

150 Looks for entry points in the "little_loops.extensions" group. 

151 

152 Returns: 

153 List of successfully loaded extension instances 

154 """ 

155 extensions: list[Any] = [] 

156 try: 

157 eps = entry_points(group=ENTRY_POINT_GROUP) 

158 except TypeError: 

159 # Python 3.11 compatibility: entry_points() may not support group kwarg 

160 eps = entry_points().get(ENTRY_POINT_GROUP, []) # type: ignore[attr-defined, assignment] 

161 for ep in eps: 

162 try: 

163 cls = ep.load() 

164 extensions.append(cls()) 

165 except Exception: 

166 logger.warning("Failed to load extension entry point: %s", ep.name, exc_info=True) 

167 return extensions 

168 

169 @staticmethod 

170 def load_all(config_paths: list[str] | None = None) -> list[LLExtension]: 

171 """Load extensions from all discovery sources. 

172 

173 Combines extensions from config paths and entry points. 

174 

175 Args: 

176 config_paths: Optional list of "module:Class" strings from config 

177 

178 Returns: 

179 Combined list of all loaded extensions 

180 """ 

181 extensions: list[Any] = [] 

182 if config_paths: 

183 extensions.extend(ExtensionLoader.from_config(config_paths)) 

184 extensions.extend(ExtensionLoader.from_entry_points()) 

185 return extensions 

186 

187 

188def wire_extensions( 

189 bus: EventBus, 

190 config_paths: list[str] | None = None, 

191 executor: FSMExecutor | PersistentExecutor | None = None, 

192) -> list[LLExtension]: 

193 """Load extensions and register them on an EventBus and optional FSMExecutor. 

194 

195 Each extension's ``on_event`` callback is wrapped to convert the raw 

196 ``dict[str, Any]`` dispatched by ``EventBus.emit()`` into an ``LLEvent`` 

197 using ``from_raw_event()`` (which copies the dict to avoid mutation). 

198 

199 When ``executor`` is provided, a second pass populates 

200 ``executor._contributed_actions``, ``executor._contributed_evaluators``, 

201 and ``executor._interceptors`` from each extension that implements the 

202 corresponding protocols. 

203 

204 Args: 

205 bus: EventBus to register extension callbacks on 

206 config_paths: Optional list of "module:Class" strings from config 

207 executor: Optional FSMExecutor to populate with contributed types 

208 

209 Returns: 

210 List of loaded extension instances 

211 """ 

212 extensions = ExtensionLoader.load_all(config_paths) 

213 extensions = sorted(extensions, key=lambda e: getattr(e, "priority", 0)) 

214 

215 def _make_callback(e: LLExtension) -> EventCallback: 

216 def _cb(event: dict[str, Any]) -> None: 

217 e.on_event(LLEvent.from_raw_event(event)) 

218 

219 return _cb 

220 

221 for ext in extensions: 

222 if hasattr(ext, "on_event"): 

223 bus.register(_make_callback(ext), filter=getattr(ext, "event_filter", None)) 

224 

225 fsm_executor: FSMExecutor | None 

226 if executor is None: 

227 fsm_executor = None 

228 elif hasattr(executor, "_executor"): 

229 fsm_executor = executor._executor 

230 else: 

231 fsm_executor = executor 

232 

233 if fsm_executor is not None: 

234 for ext in extensions: 

235 if hasattr(ext, "provided_actions"): 

236 for name in ext.provided_actions(): 

237 if name in fsm_executor._contributed_actions: 

238 raise ValueError( 

239 f"Extension conflict: action '{name}' already registered by another extension" 

240 ) 

241 fsm_executor._contributed_actions.update(ext.provided_actions()) 

242 if hasattr(ext, "provided_evaluators"): 

243 for name in ext.provided_evaluators(): 

244 if name in fsm_executor._contributed_evaluators: 

245 raise ValueError( 

246 f"Extension conflict: evaluator '{name}' already registered by another extension" 

247 ) 

248 fsm_executor._contributed_evaluators.update(ext.provided_evaluators()) 

249 if ( 

250 hasattr(ext, "before_route") 

251 or hasattr(ext, "after_route") 

252 or hasattr(ext, "before_issue_close") 

253 ): 

254 fsm_executor._interceptors.append(ext) 

255 

256 if extensions: 

257 logger.info("Wired %d extension(s) to EventBus", len(extensions)) 

258 return extensions