Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-skills/src/lexigram/ai/skills/di/provider.py: 52%

1"""SkillsProvider — DI provider registering the skills subsystem.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from typing import TYPE_CHECKING, Any, cast 

7 

8from lexigram.ai.skills.config import SkillsConfig 

9from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

10from lexigram.contracts.core.provider import ProviderPriority 

11from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol 

12from lexigram.di.provider import Provider 

13from lexigram.logging import ( 

14 get_logger, 

15) 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.core.di import ( 

19 ContainerRegistrarProtocol, 

20 ContainerResolverProtocol, 

21 ) 

22 

23logger = get_logger(__name__) 

24 

25_BUILTIN_MAP: dict[str, str] = { 

26 "current_datetime": "lexigram.ai.skills.builtin.datetime_skill.DateTimeSkill", 

27 "math_calculate": "lexigram.ai.skills.builtin.math_skill.MathSkill", 

28 "text_summarize": "lexigram.ai.skills.builtin.text_processing.TextSummarizeSkill", 

29 "text_translate": "lexigram.ai.skills.builtin.text_processing.TextTranslateSkill", 

30 "web_search": "lexigram.ai.skills.builtin.web_search.WebSearchSkill", 

31 "http_request": "lexigram.ai.skills.builtin.http_request.HTTPRequestSkill", 

32 "file_read": "lexigram.ai.skills.builtin.file_operations.FileReadSkill", 

33 "file_write": "lexigram.ai.skills.builtin.file_operations.FileWriteSkill", 

34 "code_execute": "lexigram.ai.skills.builtin.code_execution.CodeExecutionSkill", 

35} 

36 

37 

38def _import_class(dotted_path: str) -> type[Any]: 

39 """Import a class from a fully qualified dotted path.""" 

40 import importlib 

41 

42 module_path, _, class_name = dotted_path.rpartition(".") 

43 module = importlib.import_module(module_path) 

44 return getattr(module, class_name) 

45 

46 

47class SkillsProvider(Provider): 

48 """Register and bootstrap the skills subsystem.""" 

49 

50 name = "skills" 

51 priority = ProviderPriority.DOMAIN 

52 config_key: str | None = "ai_skills" 

53 config_model: type | None = SkillsConfig 

54 

55 def __init__( 

56 self, 

57 config: SkillsConfig | None = None, 

58 enable_tool_bridge: bool = False, 

59 **kwargs: Any, 

60 ) -> None: 

61 """Initialise the provider with optional configuration. 

62 

63 Args: 

64 config: Skills configuration. Defaults to ``SkillsConfig()``. 

65 enable_tool_bridge: Register the tool-bridge adapter so agent tool 

66 registries can invoke skills as first-class tools. 

67 **kwargs: Reserved for future keyword arguments. 

68 """ 

69 super().__init__() 

70 self._requested_config = config 

71 self._config = config or SkillsConfig() 

72 self._enable_tool_bridge = enable_tool_bridge 

73 

74 async def register(self, container: ContainerRegistrarProtocol) -> None: 

75 """Register all skills services into the container.""" 

76 from lexigram.ai.skills.caching.skill_cache import SkillResultCache 

77 from lexigram.ai.skills.executor import SkillExecutor 

78 from lexigram.ai.skills.permissions.permission_checker import PermissionChecker 

79 from lexigram.ai.skills.registry import SkillRegistry 

80 

81 self._config = self._requested_config or ( 

82 self.config 

83 if isinstance(getattr(self, "config", None), SkillsConfig) 

84 else self._config 

85 ) 

86 container.singleton(SkillsConfig, instance=self._config) 

87 registry = SkillRegistry() 

88 container.singleton(SkillRegistry, instance=registry) 

89 logger.debug("skills_provider_registered", service="SkillRegistry") 

90 

91 cache: SkillResultCache | None = None 

92 if self._config.cache_enabled: 

93 cache = SkillResultCache(ttl_seconds=self._config.cache_ttl_seconds) 

94 container.singleton(SkillResultCache, instance=cache) 

95 logger.debug("skills_provider_registered", service="SkillResultCache") 

96 

97 permission_checker: PermissionChecker | None = None 

98 if self._config.enforce_permissions: 

99 permission_checker = PermissionChecker() 

100 container.singleton(PermissionChecker, instance=permission_checker) 

101 logger.debug("skills_provider_registered", service="PermissionChecker") 

102 

103 executor = SkillExecutor( 

104 registry=registry, 

105 cache=cast("CacheBackendProtocol | None", cache), 

106 permission_checker=permission_checker, 

107 ) 

108 container.singleton(SkillExecutor, instance=executor) 

109 logger.debug("skills_provider_registered", service="SkillExecutor") 

110 

111 async def boot(self, container: ContainerResolverProtocol) -> None: 

112 """Register built-in and auto-discovered skills into the registry.""" 

113 from lexigram.ai.skills.registry import SkillRegistry 

114 

115 registry = await container.resolve_optional(SkillRegistry) 

116 if registry is None: 

117 logger.warning("skills_registry_not_available") 

118 return 

119 

120 if self._config.enable_builtin: 

121 for skill_name in self._config.builtin_skills: 

122 dotted = _BUILTIN_MAP.get(skill_name) 

123 if dotted is None: 

124 logger.warning("skills_unknown_builtin", skill=skill_name) 

125 continue 

126 try: 

127 cls = _import_class(dotted) 

128 registry.register(cls()) 

129 logger.debug("skills_builtin_registered", skill=skill_name) 

130 except ( 

131 ImportError, 

132 AttributeError, 

133 TypeError, 

134 ValueError, 

135 ) as exc: 

136 logger.warning( 

137 "skills_builtin_register_error", 

138 skill=skill_name, 

139 error=str(exc), 

140 ) 

141 

142 if self._config.auto_discover: 

143 from lexigram.ai.skills.discovery.module_scanner import ModuleScanner 

144 

145 scanner = ModuleScanner() 

146 for pkg in self._config.scan_packages: 

147 try: 

148 count = await scanner.scan_package(registry, pkg) 

149 logger.info("skills_auto_discovered", package=pkg, count=count) 

150 except ( 

151 ImportError, 

152 AttributeError, 

153 RuntimeError, 

154 ValueError, 

155 ) as exc: 

156 logger.warning( 

157 "skills_discovery_error", package=pkg, error=str(exc) 

158 ) 

159 

160 # Skill sources (SKILL.md files from various directories) 

161 if self._config.enable_skill_sources: 

162 from lexigram.ai.skills.discovery.skill_loader import SkillLoader 

163 from lexigram.ai.skills.discovery.skill_source_scanner import ( 

164 SkillSourceScanner, 

165 ) 

166 

167 # Scan all configured paths 

168 for path_str in self._config.skill_paths: 

169 path = Path(path_str).expanduser() 

170 if not path.exists(): 

171 logger.debug("skill_source_path_not_found", path=path_str) 

172 continue 

173 

174 loader = SkillLoader( 

175 timeout_seconds=self._config.script_timeout_seconds, 

176 skill_root=path, 

177 allowed_script_types=tuple(self._config.allowed_script_types), 

178 ) 

179 source_scanner = SkillSourceScanner(loader=loader) 

180 

181 try: 

182 count = await source_scanner.scan(registry, str(path)) 

183 logger.info("skill_sources_discovered", path=path_str, count=count) 

184 except Exception as exc: # noqa: BLE001 # skill source scanning is best-effort; one failure must not stop others 

185 logger.warning( 

186 "skill_source_discovery_error", 

187 path=path_str, 

188 error=str(exc), 

189 ) 

190 

191 async def shutdown(self) -> None: 

192 """Perform any cleanup on shutdown (no-op for the skills provider).""" 

193 

194 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

195 """Health check — always healthy (in-process domain provider). 

196 

197 No external backend to ping. 

198 

199 Args: 

200 timeout: Ignored for in-process providers. 

201 

202 Returns: 

203 Always HEALTHY — no external backend to ping. 

204 """ 

205 return HealthCheckResult( 

206 component=self.name, 

207 status=HealthStatus.HEALTHY, 

208 details={"status": "operational"}, 

209 ) 

210 

211 

212__all__ = ["SkillsProvider"]