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

1"""Governance DI provider.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, cast 

6 

7from lexigram.ai.governance.config import GovernanceConfig 

8from lexigram.ai.governance.relay_billing.di import ( 

9 boot_relay_billing, 

10 register_relay_billing, 

11) 

12from lexigram.ai.governance.relay_channels.di import ( 

13 boot_relay_channels, 

14 register_relay_channels, 

15) 

16from lexigram.ai.governance.relay_ledger.di import ( 

17 boot_relay_ledger, 

18 register_relay_ledger, 

19) 

20from lexigram.ai.governance.relay_logs.di import ( 

21 boot_relay_logs, 

22 register_relay_logs, 

23) 

24from lexigram.ai.governance.services.manager import AIGovernanceManager 

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

26from lexigram.contracts.core.provider import ProviderPriority 

27from lexigram.di.provider import Provider 

28from lexigram.logging import ( 

29 get_logger, 

30) 

31 

32if TYPE_CHECKING: 

33 from lexigram.contracts.core.di import ( 

34 BootContainerProtocol, 

35 ContainerRegistrarProtocol, 

36 ContainerResolverProtocol, 

37 ) 

38 

39logger = get_logger(__name__) 

40 

41 

42class GovernanceProvider(Provider): 

43 """Provider for AI Governance. 

44 

45 Registers :class:`~lexigram.ai.governance.services.manager.AIGovernanceManager`. 

46 """ 

47 

48 name = "governance" 

49 priority = ProviderPriority.DOMAIN 

50 config_key: str | None = "ai_governance" 

51 config_model: type | None = GovernanceConfig 

52 

53 def __init__( 

54 self, 

55 config: GovernanceConfig | dict | None = None, 

56 ) -> None: 

57 super().__init__() 

58 if isinstance(config, dict): 

59 config = GovernanceConfig(**config) 

60 elif config is not None and not isinstance(config, GovernanceConfig): 

61 raise TypeError( 

62 f"config must be GovernanceConfig or dict, got {type(config).__name__}" 

63 ) 

64 self._requested_config = config 

65 self._config = config or GovernanceConfig() 

66 

67 @classmethod 

68 def from_config( 

69 cls, 

70 config: GovernanceConfig, 

71 **context: object, 

72 ) -> GovernanceProvider: 

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

74 del context 

75 return cls(config) 

76 

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

78 """Register the governance services.""" 

79 from lexigram.contracts.ai.governance import AIGovernanceProtocol 

80 

81 self._config = self._requested_config or ( 

82 self.config 

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

84 else self._config 

85 ) 

86 container.singleton(GovernanceConfig, self._config) 

87 register_relay_billing(container, self._config) 

88 register_relay_logs(container, self._config) 

89 register_relay_channels(container, self._config) 

90 register_relay_ledger(container, self._config) 

91 

92 if not self._config.enabled: 

93 logger.info("governance_disabled", reason="GovernanceConfig.enabled=False") 

94 return 

95 

96 from lexigram.ai.governance.audit import AIAuditStore, InMemoryAuditStore 

97 

98 audit_store: AIAuditStore = InMemoryAuditStore() 

99 container.singleton(AIAuditStore, audit_store) 

100 

101 manager = AIGovernanceManager(self._config, audit_store=audit_store) 

102 container.singleton(AIGovernanceManager, manager) 

103 container.singleton(AIGovernanceProtocol, manager) 

104 

105 # Register resource unit registry + tracker when units are configured. 

106 # Reuse the manager's internal tracker so consume/release routes the 

107 # same backend regardless of which protocol is resolved. 

108 if self._config.resource_units and manager.resource_tracker is not None: 

109 from lexigram.ai.governance.resource.registry import ( 

110 ResourceUnitRegistry, 

111 ) 

112 from lexigram.ai.governance.resource.tracker import ( 

113 ResourceUnitTracker, 

114 ) 

115 

116 assert manager._resource_registry is not None # noqa: S101 

117 container.singleton(ResourceUnitRegistry, manager._resource_registry) 

118 container.singleton(ResourceUnitTracker, manager.resource_tracker) 

119 logger.info( 

120 "resource_units_registered", 

121 count=len(self._config.resource_units), 

122 ) 

123 

124 logger.info("governance_registered") 

125 

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

127 """Attach governance persistence and boot remaining sub-systems. 

128 

129 The persistence backend is resolved here, after container freezing, 

130 so that the optional database/cache backends are available and the 

131 persistence-aware manager can be rebound (spec: DI ordering). 

132 

133 Args: 

134 container: The DI container. 

135 """ 

136 if not self._config.enabled: 

137 return 

138 

139 from lexigram.ai.governance.audit import AIAuditStore 

140 from lexigram.contracts import CacheBackendProtocol, DatabaseProviderProtocol 

141 from lexigram.contracts.ai.governance import AIGovernanceProtocol 

142 

143 manager = await container.resolve(AIGovernanceManager) 

144 audit_store = await container.resolve_optional(AIAuditStore) 

145 

146 database = await container.resolve_optional(DatabaseProviderProtocol) 

147 cache = await container.resolve_optional(CacheBackendProtocol) 

148 

149 if database is not None: 

150 from lexigram.ai.governance.persistence import ( 

151 DatabaseGovernancePersistence, 

152 GovernancePersistence, 

153 ) 

154 

155 persistence = cast( 

156 "GovernancePersistence", DatabaseGovernancePersistence(database) 

157 ) 

158 manager_with_persistence = AIGovernanceManager( 

159 self._config, persistence=persistence, audit_store=audit_store 

160 ) 

161 elif cache is not None: 

162 from lexigram.ai.governance.persistence import ( 

163 GovernancePersistence, 

164 RedisGovernancePersistence, 

165 ) 

166 

167 persistence = cast( 

168 "GovernancePersistence", RedisGovernancePersistence(cache) 

169 ) 

170 manager_with_persistence = AIGovernanceManager( 

171 self._config, persistence=persistence, audit_store=audit_store 

172 ) 

173 else: 

174 logger.info( 

175 "governance_persistence_skip", 

176 reason="no database or cache backend available", 

177 backend="in-memory", 

178 ) 

179 manager_with_persistence = manager 

180 

181 if manager_with_persistence is not manager: 

182 boot_container = cast("BootContainerProtocol", container) 

183 boot_container.bind(AIGovernanceManager, manager_with_persistence) 

184 boot_container.bind(AIGovernanceProtocol, manager_with_persistence) 

185 

186 # Keep the resource-unit tracker/registry routing through the 

187 # same backend as the rebound manager (see register()). 

188 if ( 

189 self._config.resource_units 

190 and manager_with_persistence.resource_tracker is not None 

191 ): 

192 from lexigram.ai.governance.resource.registry import ( 

193 ResourceUnitRegistry, 

194 ) 

195 from lexigram.ai.governance.resource.tracker import ( 

196 ResourceUnitTracker, 

197 ) 

198 

199 boot_container.bind( 

200 ResourceUnitRegistry, manager_with_persistence._resource_registry 

201 ) 

202 boot_container.bind( 

203 ResourceUnitTracker, manager_with_persistence.resource_tracker 

204 ) 

205 

206 await boot_relay_billing( 

207 cast("BootContainerProtocol", container), 

208 self._config, 

209 ) 

210 await boot_relay_logs( 

211 cast("BootContainerProtocol", container), 

212 self._config, 

213 ) 

214 await boot_relay_channels( 

215 cast("BootContainerProtocol", container), 

216 self._config, 

217 ) 

218 await boot_relay_ledger( 

219 cast("BootContainerProtocol", container), 

220 self._config, 

221 ) 

222 logger.debug("governance_booted") 

223 

224 async def shutdown(self) -> None: 

225 """Shutdown phase.""" 

226 logger.debug("governance_shutdown") 

227 

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

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

230 

231 No external backend to ping. 

232 

233 Args: 

234 timeout: Ignored for in-process providers. 

235 

236 Returns: 

237 Always HEALTHY — no external backend to ping. 

238 """ 

239 return HealthCheckResult( 

240 component=self.name, 

241 status=HealthStatus.HEALTHY, 

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

243 ) 

244 

245 

246__all__ = ["GovernanceProvider"]