Coverage for session_buddy / utils / instance_managers.py: 65.77%

1"""Instance managers for MCP server singletons. 

2 

3This module provides lazy initialization and access to global singleton instances 

4for application monitoring, LLM providers, and serverless session management. 

5 

6Extracted from server.py Phase 2.6 to reduce cognitive complexity. 

7""" 

8 

9from __future__ import annotations 

10 

11import os 

12from contextlib import suppress 

13from pathlib import Path 

14from typing import TYPE_CHECKING, Any 

15 

16from session_buddy.di import SessionPaths, get_sync_typed 

17from session_buddy.di.container import depends 

18 

19if TYPE_CHECKING: 

20 from session_buddy.adapters.reflection_adapter import ( 

21 ReflectionDatabaseAdapter as ReflectionDatabase, 

22 ) 

23 from session_buddy.app_monitor import ApplicationMonitor 

24 from session_buddy.interruption_manager import InterruptionManager 

25 from session_buddy.llm_providers import LLMManager 

26 from session_buddy.serverless_mode import ServerlessSessionManager 

27 

28 

29async def get_app_monitor() -> ApplicationMonitor | None: 

30 """Resolve application monitor via DI, creating it on demand. 

31 

32 Note: 

33 Uses the Oneiric-backed service container for singleton resolution. 

34 

35 """ 

36 try: 

37 from session_buddy.app_monitor import ApplicationMonitor 

38 except ImportError: 

39 return None 

40 

41 with suppress(Exception): 

42 monitor = get_sync_typed(ApplicationMonitor) # type: ignore[no-any-return] 

43 if isinstance(monitor, ApplicationMonitor): 43 ↛ 46line 43 didn't jump to line 46

44 return monitor 

45 

46 data_dir = _resolve_claude_dir() / "data" / "app_monitoring" 

47 working_dir = Path(os.environ.get("PWD", str(Path.cwd()))) 

48 project_paths = [str(working_dir)] if working_dir.exists() else [] 

49 

50 monitor = ApplicationMonitor(str(data_dir), project_paths) 

51 depends.set(ApplicationMonitor, monitor) 

52 return monitor 

53 

54 

55async def get_llm_manager() -> LLMManager | None: 

56 """Resolve LLM manager via DI, creating it on demand. 

57 

58 Note: 

59 Uses the Oneiric-backed service container for singleton resolution. 

60 

61 """ 

62 try: 

63 from session_buddy.llm_providers import LLMManager 

64 except ImportError: 

65 return None 

66 

67 with suppress(Exception): 

68 manager = get_sync_typed(LLMManager) # type: ignore[no-any-return] 

69 if isinstance(manager, LLMManager): 69 ↛ 72line 69 didn't jump to line 72

70 return manager 

71 

72 config_path = _resolve_claude_dir() / "data" / "llm_config.json" 

73 manager = LLMManager(str(config_path) if config_path.exists() else None) 

74 depends.set(LLMManager, manager) 

75 return manager 

76 

77 

78async def get_serverless_manager() -> ServerlessSessionManager | None: 

79 """Resolve serverless session manager via DI, creating it on demand. 

80 

81 Note: 

82 Uses the Oneiric-backed service container for singleton resolution. 

83 

84 """ 

85 try: 

86 from session_buddy.serverless_mode import ( 

87 ServerlessConfigManager, 

88 ServerlessSessionManager, 

89 ) 

90 except ImportError: 

91 return None 

92 

93 with suppress(Exception): 

94 manager = get_sync_typed(ServerlessSessionManager) # type: ignore[no-any-return] 

95 if isinstance(manager, ServerlessSessionManager): 95 ↛ 98line 95 didn't jump to line 98

96 return manager 

97 

98 claude_dir = _resolve_claude_dir() 

99 config_path = claude_dir / "data" / "serverless_config.json" 

100 config = ServerlessConfigManager.load_config( 

101 str(config_path) if config_path.exists() else None, 

102 ) 

103 storage_backend = ServerlessConfigManager.create_storage_backend(config) 

104 manager = ServerlessSessionManager(storage_backend) 

105 depends.set(ServerlessSessionManager, manager) 

106 return manager 

107 

108 

109async def get_reflection_database() -> ReflectionDatabase | None: 

110 """Resolve reflection database via DI, creating it on demand. 

111 

112 Note: 

113 Returns ReflectionDatabaseAdapter which maintains API compatibility 

114 with the original ReflectionDatabase while using Oneiric DuckDB. 

115 

116 """ 

117 try: 

118 from session_buddy.adapters.reflection_adapter_oneiric import ( 

119 ReflectionDatabaseAdapterOneiric as ReflectionDatabaseAdapter, 

120 ) 

121 except ImportError: 

122 try: 

123 from session_buddy.adapters.reflection_adapter import ( 

124 ReflectionDatabaseAdapter, 

125 ) 

126 except ImportError: 

127 return None 

128 

129 # Note: We use ReflectionDatabaseAdapter as the key for the new implementation 

130 with suppress(Exception): 

131 db = depends.get_sync(ReflectionDatabaseAdapter) 

132 if isinstance(db, ReflectionDatabaseAdapter): 132 ↛ 135line 132 didn't jump to line 135

133 return db 

134 

135 from session_buddy.adapters.lifecycle import init_reflection_adapter 

136 

137 await init_reflection_adapter() 

138 with suppress(Exception): 

139 db = depends.get_sync(ReflectionDatabaseAdapter) 

140 if isinstance(db, ReflectionDatabaseAdapter): 140 ↛ 142line 140 didn't jump to line 142

141 return db 

142 return None 

143 

144 

145async def get_interruption_manager() -> InterruptionManager | None: 

146 """Resolve interruption manager via DI, creating it on demand. 

147 

148 Note: 

149 Uses the Oneiric-backed service container for singleton resolution. 

150 

151 """ 

152 try: 

153 from session_buddy.interruption_manager import InterruptionManager 

154 except ImportError: 

155 return None 

156 

157 with suppress(Exception): 

158 manager = get_sync_typed(InterruptionManager) # type: ignore[no-any-return] 

159 if isinstance(manager, InterruptionManager): 

160 return manager 

161 

162 manager = InterruptionManager() 

163 depends.set(InterruptionManager, manager) 

164 return manager 

165 

166 

167def reset_instances() -> None: 

168 """Reset registered instances in the DI container.""" 

169 depends.reset() 

170 

171 

172def _resolve_claude_dir() -> Path: 

173 """Resolve claude directory via type-safe DI. 

174 

175 Returns: 

176 Path to .claude directory, using SessionPaths from DI container 

177 or falling back to default home directory. 

178 

179 Note: 

180 Uses SessionPaths type for DI resolution instead of string keys. 

181 

182 """ 

183 with suppress(KeyError, AttributeError, RuntimeError, TypeError): 

184 # RuntimeError: when adapter requires async 

185 # TypeError: when DI has type confusion 

186 paths = depends.get_sync(SessionPaths) 

187 if isinstance(paths, SessionPaths): 187 ↛ 192line 187 didn't jump to line 192

188 paths.claude_dir.mkdir(parents=True, exist_ok=True) 

189 return paths.claude_dir 

190 

191 # Fallback: create default paths if not registered 

192 default_dir = Path(os.path.expanduser("~")) / ".claude" 

193 default_dir.mkdir(parents=True, exist_ok=True) 

194 return default_dir 

195 

196 

197def _iter_dependencies() -> list[type[Any]]: 

198 deps: list[type[Any]] = [] 

199 with suppress(ImportError): 

200 from session_buddy.app_monitor import ApplicationMonitor 

201 

202 deps.append(ApplicationMonitor) 

203 with suppress(ImportError): 

204 from session_buddy.llm_providers import LLMManager 

205 

206 deps.append(LLMManager) 

207 with suppress(ImportError): 

208 from session_buddy.interruption_manager import InterruptionManager 

209 

210 deps.append(InterruptionManager) 

211 with suppress(ImportError): 

212 from session_buddy.serverless_mode import ServerlessSessionManager 

213 

214 deps.append(ServerlessSessionManager) 

215 with suppress(ImportError): 

216 from session_buddy.adapters.reflection_adapter import ( 

217 ReflectionDatabaseAdapter, 

218 ) 

219 

220 deps.append(ReflectionDatabaseAdapter) 

221 return deps