MCPcopy Create free account
hub / github.com/agentforce314/clawcodex / HookConfigManager

Class HookConfigManager

src/hooks/config_manager.py:302–374  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

300
301
302class HookConfigManager:
303 def __init__(
304 self,
305 registry: AsyncHookRegistry,
306 settings_path: str | Path | None = None,
307 ) -> None:
308 self._registry = registry
309 self._settings_path = Path(settings_path) if settings_path else _get_settings_path()
310 self._snapshot: HookConfigSnapshot | None = None
311 self._last_mtime: float = 0.0
312
313 @property
314 def snapshot(self) -> HookConfigSnapshot | None:
315 return self._snapshot
316
317 async def load(self) -> HookConfigSnapshot:
318 snapshot = load_hooks_from_settings(self._settings_path)
319 self._snapshot = snapshot
320
321 await self._registry.clear_source(HookSource.USER_SETTINGS)
322
323 for event_name, hook_configs in snapshot.hooks.items():
324 for config in hook_configs:
325 if event_name in ALL_HOOK_EVENTS:
326 await self._registry.register(
327 event_name, # type: ignore[arg-type]
328 config,
329 HookSource.USER_SETTINGS,
330 )
331
332 try:
333 self._last_mtime = self._settings_path.stat().st_mtime
334 except OSError:
335 self._last_mtime = 0.0
336
337 return snapshot
338
339 async def reload_if_changed(self) -> bool:
340 try:
341 current_mtime = self._settings_path.stat().st_mtime
342 except OSError:
343 return False
344
345 if current_mtime > self._last_mtime:
346 await self.load()
347 return True
348
349 return False
350
351 async def validate(self) -> list[HookValidationError]:
352 if not self._settings_path.exists():
353 return []
354
355 try:
356 data = json.loads(self._settings_path.read_text(encoding="utf-8"))
357 except (json.JSONDecodeError, OSError):
358 return [HookValidationError(
359 event="",

Calls

no outgoing calls