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

1"""Observability DI provider.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.ai.observability.config import ObservabilityConfig 

8from lexigram.ai.observability.health import AIHealthMonitor 

9from lexigram.ai.observability.metrics import AIMetrics 

10from lexigram.ai.observability.tracing import AITracer 

11from lexigram.contracts.ai import LLMClientProtocol 

12from lexigram.contracts.ai.governance import AIAuditStoreProtocol 

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

14from lexigram.contracts.core.provider import ProviderPriority 

15from lexigram.contracts.data.vector.protocols import VectorStoreProtocol 

16from lexigram.contracts.exceptions.container import UnresolvableDependencyError 

17from lexigram.contracts.observability.ai import ( 

18 AIHealthMonitorProtocol, 

19 AIMetricsProtocol, 

20 AITracerProtocol, 

21) 

22from lexigram.di.provider import Provider 

23from lexigram.logging import ( 

24 get_logger, 

25) 

26from lexigram.logging.redaction import DefaultRedactor 

27 

28if TYPE_CHECKING: 

29 from lexigram.contracts.core.di import ( 

30 BootContainerProtocol, 

31 ContainerRegistrarProtocol, 

32 ) 

33 

34logger = get_logger(__name__) 

35 

36 

37class ObservabilityProvider(Provider): 

38 """Provider for AI Observability. 

39 

40 Registers :class:`~lexigram.ai.observability.metrics.AIMetrics`, 

41 :class:`~lexigram.ai.observability.tracing.AITracer`, and 

42 :class:`~lexigram.ai.observability.health.AIHealthMonitor`. 

43 

44 During ``boot()``, self-wires observability decorators around any 

45 ``LLMClientProtocol`` and ``VectorStoreProtocol`` that are already 

46 registered in the container, so the wrapping is transparent to callers. 

47 """ 

48 

49 name = "ai-observability" 

50 priority = ProviderPriority.DOMAIN 

51 config_key: str | None = "ai_observability" 

52 config_model: type | None = ObservabilityConfig 

53 

54 def __init__(self, config: ObservabilityConfig | None = None) -> None: 

55 super().__init__() 

56 self._requested_config = config 

57 self._config = config or ObservabilityConfig() 

58 

59 @classmethod 

60 def from_config( 

61 cls, config: ObservabilityConfig, **context 

62 ) -> ObservabilityProvider: 

63 """Factory method for DI container setup.""" 

64 return cls(config) 

65 

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

67 """Register the observability services.""" 

68 self._config = self._requested_config or ( 

69 self.config 

70 if isinstance(getattr(self, "config", None), ObservabilityConfig) 

71 else self._config 

72 ) 

73 container.singleton(ObservabilityConfig, self._config) 

74 

75 if not self._config.enabled: 

76 logger.info( 

77 "observability_disabled", reason="ObservabilityConfig.enabled=False" 

78 ) 

79 return 

80 

81 # Register singletons for the core observability classes. 

82 # Also register under the contracts protocols so callers resolved 

83 # by protocol receive the same instances. 

84 container.singleton(AIMetrics) 

85 container.singleton(AIMetricsProtocol, AIMetrics) 

86 redaction_policy = ( 

87 DefaultRedactor() if self._config.trace_redaction_enabled else None 

88 ) 

89 max_attribute_length = ( 

90 self._config.trace_max_attribute_length 

91 if self._config.trace_max_attribute_length > 0 

92 else None 

93 ) 

94 if redaction_policy is None and max_attribute_length is None: 

95 container.singleton(AITracer) 

96 container.singleton(AITracerProtocol, AITracer) 

97 else: 

98 tracer_instance = AITracer( 

99 redaction_policy=redaction_policy, 

100 max_attribute_length=max_attribute_length, 

101 ) 

102 container.singleton(AITracer, tracer_instance) 

103 container.singleton(AITracerProtocol, tracer_instance) 

104 container.singleton(AIHealthMonitor) 

105 container.singleton(AIHealthMonitorProtocol, AIHealthMonitor) 

106 

107 logger.info("observability_registered") 

108 

109 async def boot(self, container: BootContainerProtocol) -> None: 

110 """Boot phase — self-wire observability wrappers into the container. 

111 

112 If ``LLMClientProtocol`` or ``VectorStoreProtocol`` are registered, 

113 they are replaced with instrumented wrappers. Both ``AITracer`` and 

114 ``AIMetrics`` must be available; if either is missing the wrapping is 

115 skipped gracefully. 

116 """ 

117 if not self._config.enabled: 

118 logger.debug("observability_booted_skipped", reason="disabled") 

119 return 

120 

121 tracer: AITracer | None = None 

122 metrics: AIMetrics | None = None 

123 audit_store: Any | None = None 

124 

125 try: 

126 tracer = await container.resolve(AITracer) 

127 metrics = await container.resolve(AIMetrics) 

128 except (ValueError, KeyError, TypeError): 

129 logger.debug("observability_tracer_metrics_unavailable") 

130 

131 try: 

132 audit_store = await container.resolve(AIAuditStoreProtocol) 

133 except (ValueError, KeyError, TypeError, UnresolvableDependencyError): 

134 pass # audit store is optional 

135 

136 # Wrap LLM client if registered 

137 if tracer or metrics: 

138 try: 

139 raw_llm = await container.resolve(LLMClientProtocol) 

140 from lexigram.ai.observability.wrappers import ObservableLLMClient 

141 

142 observable_llm = ObservableLLMClient( 

143 raw_llm, 

144 provider=getattr(raw_llm, "_provider", "unknown"), 

145 model=getattr(raw_llm, "_model", "unknown"), 

146 tracer=tracer, 

147 metrics=metrics, 

148 audit_store=audit_store, 

149 redaction_policy=( 

150 DefaultRedactor() 

151 if self._config.trace_redaction_enabled 

152 else None 

153 ), 

154 ) 

155 container.bind(LLMClientProtocol, observable_llm) 

156 logger.info("observability_llm_wrapped") 

157 except (ValueError, KeyError, TypeError, UnresolvableDependencyError): 

158 logger.debug("observability_no_llm_to_wrap") 

159 

160 # Wrap vector store if registered 

161 try: 

162 raw_store = await container.resolve(VectorStoreProtocol) 

163 from lexigram.ai.observability.wrappers import ObservableVectorStore 

164 

165 observable_store = ObservableVectorStore( 

166 raw_store, 

167 backend=getattr(raw_store, "_backend", "unknown"), 

168 collection=getattr(raw_store, "_collection", "unknown"), 

169 tracer=tracer, 

170 metrics=metrics, 

171 ) 

172 container.bind(VectorStoreProtocol, observable_store) 

173 logger.info("observability_vector_wrapped") 

174 except (ValueError, KeyError, TypeError, UnresolvableDependencyError): 

175 logger.debug("observability_no_vector_to_wrap") 

176 

177 logger.debug("observability_booted") 

178 

179 async def shutdown(self) -> None: 

180 """Shutdown phase.""" 

181 logger.debug("observability_shutdown") 

182 

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

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

185 

186 No external backend to ping. 

187 

188 Args: 

189 timeout: Ignored for in-process providers. 

190 

191 Returns: 

192 Always HEALTHY — no external backend to ping. 

193 """ 

194 return HealthCheckResult( 

195 component=self.name, 

196 status=HealthStatus.HEALTHY, 

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

198 ) 

199 

200 

201__all__ = ["ObservabilityProvider"]