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

1"""DI provider for lexigram-ai-session.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.ai.session.config import SessionConfig 

8from lexigram.ai.session.context.pruner import RelevanceContextPruner 

9from lexigram.ai.session.context.session_context import SessionContext 

10from lexigram.ai.session.manager import SessionCleanupScheduler, SessionManagerImpl 

11from lexigram.ai.session.stores.in_memory import InMemorySessionStore 

12from lexigram.contracts.ai.session import ( 

13 ContextPrunerProtocol, 

14 SessionContextProtocol, 

15 SessionManagerProtocol, 

16 SessionStoreProtocol, 

17) 

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

19from lexigram.contracts.core.provider import ProviderPriority 

20from lexigram.di.provider import Provider 

21from lexigram.logging import ( 

22 get_logger, 

23) 

24 

25if TYPE_CHECKING: 

26 from lexigram.contracts.core.di import ( 

27 BootContainerProtocol, 

28 ContainerRegistrarProtocol, 

29 ) 

30 

31logger = get_logger(__name__) 

32 

33 

34class SessionProvider(Provider): 

35 """Registers session management services in the DI container. 

36 

37 Wires ``SessionStoreProtocol`` and ``SessionManagerProtocol`` based on the 

38 configured backend. Background cleanup is started during ``boot()``. 

39 """ 

40 

41 name = "session" 

42 priority = ProviderPriority.INFRASTRUCTURE 

43 config_key: str | None = "ai_session" 

44 config_model: type | None = SessionConfig 

45 

46 def __init__( 

47 self, 

48 config: SessionConfig | dict | None = None, 

49 *, 

50 enable_cleanup_scheduler: bool = True, 

51 ) -> None: 

52 """Initialise the provider.""" 

53 super().__init__() 

54 if isinstance(config, dict): 

55 config = SessionConfig(**config) 

56 self._requested_config = config 

57 self._config = config or SessionConfig() 

58 self._enable_cleanup_scheduler = enable_cleanup_scheduler 

59 self._scheduler: SessionCleanupScheduler | None = None 

60 

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

62 """Register session services in the container.""" 

63 self._config = self._requested_config or ( 

64 self.config 

65 if isinstance(getattr(self, "config", None), SessionConfig) 

66 else self._config 

67 ) 

68 container.singleton(SessionConfig, instance=self._config) 

69 

70 if not self._config.enabled: 

71 logger.info("session_disabled", reason="SessionConfig.enabled=False") 

72 return 

73 

74 if self._config.backend == "in_memory": 

75 container.singleton(SessionStoreProtocol, instance=InMemorySessionStore()) 

76 elif self._config.backend == "cache": 

77 from lexigram.ai.session.stores.cache import CacheSessionStore 

78 

79 container.singleton(SessionStoreProtocol, factory=CacheSessionStore) 

80 elif self._config.backend == "database": 

81 from lexigram.ai.session.stores.database import DatabaseSessionStore 

82 

83 container.singleton(SessionStoreProtocol, factory=DatabaseSessionStore) 

84 else: 

85 logger.warning( 

86 "unknown_session_backend", 

87 backend=self._config.backend, 

88 fallback="in_memory", 

89 ) 

90 container.singleton(SessionStoreProtocol, instance=InMemorySessionStore()) 

91 

92 container.singleton(SessionManagerProtocol, factory=SessionManagerImpl) 

93 container.singleton(SessionCleanupScheduler, factory=SessionCleanupScheduler) 

94 container.transient(SessionContextProtocol, factory=SessionContext) # type: ignore[type-abstract] 

95 container.singleton(ContextPrunerProtocol, factory=RelevanceContextPruner) 

96 

97 logger.info("session_services_registered", backend=self._config.backend) 

98 

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

100 """Pre-warm session services and start the cleanup scheduler.""" 

101 if not self._config.enabled: 

102 return 

103 

104 store = await container.resolve(SessionStoreProtocol) 

105 initialize = getattr(store, "initialize", None) 

106 if callable(initialize): 

107 await initialize() 

108 

109 if self._enable_cleanup_scheduler: 

110 try: 

111 scheduler = await container.resolve(SessionCleanupScheduler) 

112 await scheduler.start() 

113 self._scheduler = scheduler 

114 except (LookupError, RuntimeError, AttributeError, OSError) as exc: 

115 logger.warning("failed_to_start_session_cleanup", error=str(exc)) 

116 

117 async def shutdown(self) -> None: 

118 """Stop the background cleanup loop.""" 

119 if self._scheduler is None: 

120 return 

121 

122 try: 

123 await self._scheduler.stop() 

124 except (RuntimeError, AttributeError, OSError) as exc: 

125 logger.warning("failed_to_stop_session_cleanup", error=str(exc)) 

126 

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

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

129 

130 No external backend to ping. 

131 

132 Args: 

133 timeout: Ignored for in-process providers. 

134 

135 Returns: 

136 Always HEALTHY — no external backend to ping. 

137 """ 

138 return HealthCheckResult( 

139 component=self.name, 

140 status=HealthStatus.HEALTHY, 

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

142 ) 

143 

144 

145__all__ = ["SessionProvider"]