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

1"""Prompt DI provider — registers the PromptRegistry and PromptService with the container.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from typing import TYPE_CHECKING, Any 

7 

8from lexigram.ai.prompt.assembly.assembler import CacheAwarePromptAssembler 

9from lexigram.ai.prompt.assembly.cache_strategies import ProviderCacheStrategyRegistry 

10from lexigram.ai.prompt.config import PromptConfig 

11from lexigram.ai.prompt.registry.registry import PromptRegistry 

12from lexigram.ai.prompt.rendering.sanitizer import InputSanitizer 

13from lexigram.ai.prompt.service.loader import DictPromptLoader, DirectoryPromptLoader 

14from lexigram.ai.prompt.service.models import PromptTemplate 

15from lexigram.ai.prompt.service.observer import PromptObserverProtocol 

16from lexigram.ai.prompt.service.service import PromptService, PromptServiceProtocol 

17from lexigram.contracts.ai.llm import PromptAssemblerProtocol 

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

19from lexigram.contracts.core.hooks import HookRegistryProtocol 

20from lexigram.contracts.core.provider import ProviderPriority 

21from lexigram.contracts.exceptions.container import UnresolvableDependencyError 

22from lexigram.contracts.exceptions.provider import ModuleVisibilityError 

23from lexigram.di.provider import Provider 

24from lexigram.logging import ( 

25 get_logger, 

26) 

27 

28if TYPE_CHECKING: 

29 from lexigram.contracts.core.di import ( 

30 ContainerRegistrarProtocol, 

31 ContainerResolverProtocol, 

32 ) 

33 

34logger = get_logger(__name__) 

35 

36 

37class PromptProvider(Provider): 

38 """Provider for the prompt registry and prompt service. 

39 

40 Reads :class:`~lexigram.ai.prompt.config.PromptConfig`, builds a 

41 :class:`~lexigram.ai.prompt.registry.registry.PromptRegistry` and a 

42 :class:`~lexigram.ai.prompt.service.service.PromptService`, and registers 

43 both as singletons. 

44 

45 The ``PromptService`` is configured via the keyword arguments: 

46 

47 * ``template_dir`` — path to a directory of YAML/JSON template files. 

48 * ``inline_templates`` — list of raw template dicts (see 

49 :class:`~lexigram.ai.prompt.service.loader.DictPromptLoader`). 

50 * ``observer`` — an object implementing 

51 :class:`~lexigram.ai.prompt.service.observer.PromptObserverProtocol`. 

52 

53 At least one of ``template_dir`` or ``inline_templates`` may be supplied; 

54 both are combined if provided. When neither is supplied the 

55 ``PromptService`` starts empty (useful in tests). 

56 """ 

57 

58 name = "prompt" 

59 priority = ProviderPriority.DOMAIN 

60 config_key: str | None = "ai_prompt" 

61 config_model: type | None = PromptConfig 

62 

63 def __init__( 

64 self, 

65 config: PromptConfig | None = None, 

66 *, 

67 template_dir: str | Path | None = None, 

68 inline_templates: list[dict[str, Any]] | None = None, 

69 observer: PromptObserverProtocol | None = None, 

70 ) -> None: 

71 """Initialize with optional config and template sources. 

72 

73 Args: 

74 config: Optional :class:`~lexigram.ai.prompt.config.PromptConfig`. 

75 template_dir: Optional path to a directory of YAML/JSON files. 

76 inline_templates: Optional list of raw template dicts. 

77 observer: Optional observability hook for ``PromptService``. 

78 """ 

79 super().__init__() 

80 self._requested_config = config 

81 self._config = config or PromptConfig() 

82 self._template_dir = Path(template_dir) if template_dir else None 

83 self._inline_templates: list[dict[str, Any]] = inline_templates or [] 

84 self._observer = observer 

85 

86 def _load_templates(self) -> list[PromptTemplate]: 

87 """Load all templates from configured sources.""" 

88 templates: list[PromptTemplate] = [] 

89 

90 if self._template_dir is not None: 

91 templates.extend( 

92 DirectoryPromptLoader( 

93 self._template_dir, default_format=self._config.default_format 

94 ).load() 

95 ) 

96 

97 if self._inline_templates: 

98 templates.extend( 

99 DictPromptLoader( 

100 self._inline_templates, default_format=self._config.default_format 

101 ).load() 

102 ) 

103 

104 return templates 

105 

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

107 """Register the PromptConfig, PromptRegistry, PromptService, and CacheAwarePromptAssembler.""" 

108 self._config = self._requested_config or ( 

109 self.config 

110 if isinstance(getattr(self, "config", None), PromptConfig) 

111 else self._config 

112 ) 

113 container.singleton(PromptConfig, self._config) 

114 

115 if not self._config.enabled: 

116 logger.info("prompt_disabled", reason="PromptConfig.enabled=False") 

117 return 

118 

119 registry = PromptRegistry() 

120 container.singleton(PromptRegistry, registry) 

121 

122 # Register PromptService with loaded templates 

123 templates = self._load_templates() 

124 sanitizer: InputSanitizer | None = None 

125 if self._config.sanitize_inputs: 

126 sanitizer = InputSanitizer(strict=self._config.strict_sanitizer) 

127 service = PromptService( 

128 templates=templates, 

129 observer=self._observer, 

130 sanitizer=sanitizer, 

131 ) 

132 container.singleton(PromptService, service) 

133 container.singleton(PromptServiceProtocol, service) 

134 

135 # Register cache-aware assembler components 

136 strategy_registry = ProviderCacheStrategyRegistry.with_defaults() 

137 container.singleton(ProviderCacheStrategyRegistry, strategy_registry) 

138 

139 # Create assembler without token counter initially — will be enhanced in boot() 

140 assembler = CacheAwarePromptAssembler( 

141 strategy_registry=strategy_registry, 

142 token_counter=None, 

143 ) 

144 container.singleton(PromptAssemblerProtocol, assembler) 

145 

146 logger.info( 

147 "prompt_provider_registered", 

148 format=self._config.default_format, 

149 sanitize_inputs=self._config.sanitize_inputs, 

150 service_templates=len(templates), 

151 ) 

152 

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

154 """Boot phase — inject optional TokenCounterProtocol and hook registry.""" 

155 hook_registry = await container.resolve_optional(HookRegistryProtocol) 

156 service = await container.resolve(PromptService) 

157 service.attach_hook_registry(hook_registry) 

158 

159 try: 

160 from lexigram.contracts.ai.llm import TokenCounterProtocol 

161 

162 token_counter = await container.resolve(TokenCounterProtocol) 

163 assembler = await container.resolve(PromptAssemblerProtocol) 

164 if isinstance(assembler, CacheAwarePromptAssembler): 

165 assembler.set_token_counter(token_counter) 

166 logger.debug("prompt_assembler_token_counter_injected") 

167 else: 

168 logger.warning( 

169 "prompt_assembler_token_counter_skipped", 

170 assembler_type=type(assembler).__name__, 

171 ) 

172 except (UnresolvableDependencyError, ModuleVisibilityError): 

173 logger.debug("prompt_assembler_token_counter_unavailable") 

174 

175 logger.debug("prompt_provider_booted") 

176 

177 async def shutdown(self) -> None: 

178 """Shutdown phase — no cleanup required for prompt registry.""" 

179 logger.debug("prompt_provider_shutdown") 

180 

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

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

183 

184 No external backend to ping. 

185 

186 Args: 

187 timeout: Ignored for in-process providers. 

188 

189 Returns: 

190 Always HEALTHY — no external backend to ping. 

191 """ 

192 return HealthCheckResult( 

193 component=self.name, 

194 status=HealthStatus.HEALTHY, 

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

196 )