Coverage for session_buddy / di / __init__.py: 66.23%

1from __future__ import annotations 

2 

3import tempfile 

4import typing as t 

5from contextlib import suppress 

6from pathlib import Path 

7 

8from session_buddy.di.container import depends 

9 

10if t.TYPE_CHECKING: 

11 from session_buddy.core import ( 

12 SessionLifecycleManager as SessionLifecycleManagerT, 

13 ) 

14 from session_buddy.core.permissions import ( 

15 SessionPermissionsManager as SessionPermissionsManagerT, 

16 ) 

17 from session_buddy.utils.logging import ( # type: ignore[attr-defined] 

18 SessionLogger as SessionLoggerT, 

19 ) 

20 

21from .config import SessionPaths 

22from .constants import CLAUDE_DIR_KEY, COMMANDS_DIR_KEY, LOGS_DIR_KEY 

23 

24_configured = False 

25 

26 

27# Type variable for use with t.cast() in legacy type contexts 

28 

29 

30def get_sync_typed[T](key: type[T]) -> T: 

31 """Type-safe wrapper for depends.get_sync. 

32 

33 This helper provides proper type information for the dependency injection 

34 container, avoiding 'no-any-return' type checker errors. 

35 

36 Args: 

37 key: The class type to retrieve from the DI container 

38 

39 Returns: 

40 The instance of type T from the DI container 

41 

42 Example: 

43 >>> from session_buddy.core.session_manager import SessionLifecycleManager 

44 >>> manager = get_sync_typed(SessionLifecycleManager) # Properly typed 

45 

46 """ 

47 result = depends.get_sync(key) 

48 # Trust the DI container - type checker will verify usage 

49 return t.cast("T", result) # Use T directly from type parameter 

50 

51 

52def configure(*, force: bool = False) -> None: 

53 """Register default dependencies for the session-buddy MCP stack. 

54 

55 This function sets up the dependency injection container with type-safe 

56 configuration and singleton instances for the session management system. 

57 

58 Args: 

59 force: If True, re-registers all dependencies even if already configured. 

60 Used primarily for testing to reset singleton state. 

61 

62 Example: 

63 >>> from session_buddy.di import configure 

64 >>> configure() # First call registers dependencies 

65 >>> configure() # Subsequent calls are no-ops unless force=True 

66 >>> configure(force=True) # Re-registers all dependencies 

67 

68 """ 

69 global _configured 

70 if _configured and not force: 

71 return 

72 

73 # Register type-safe path configuration 

74 paths = SessionPaths.from_home() 

75 paths.ensure_directories() 

76 depends.set(SessionPaths, paths) 

77 

78 # Register services with type-safe path access 

79 _register_logger(paths.logs_dir, force) 

80 _register_session_logger(paths.logs_dir, force) # Register SessionLogger 

81 _register_permissions_manager(paths.claude_dir, force) 

82 _register_lifecycle_manager(force) 

83 

84 _configured = True 

85 

86 

87def reset() -> None: 

88 """Reset dependencies to defaults.""" 

89 # Reset singleton instances that have class-level state 

90 with suppress(ImportError, AttributeError): 

91 from session_buddy.core.permissions import SessionPermissionsManager 

92 

93 SessionPermissionsManager.reset_singleton() 

94 

95 configure(force=True) 

96 

97 

98def _register_logger(logs_dir: Path, force: bool) -> None: 

99 """Register ACB logger adapter with the given logs directory. 

100 

101 Args: 

102 logs_dir: Directory for session log files 

103 force: If True, re-registers even if already registered 

104 

105 Note: 

106 This function configures ACB's logger adapter but does NOT register it 

107 in the DI container to avoid type resolution conflicts. Components 

108 should use direct logging imports instead of DI lookup. 

109 

110 """ 

111 # Skip registration entirely - we don't need to register the logger in DI 

112 # Components should use `import logging; logger = logging.getLogger(__name__)` 

113 # This avoids DI type confusion when crackerjack runs 

114 

115 

116def _register_session_logger(logs_dir: Path, force: bool) -> None: 

117 """Register SessionLogger with the given logs directory. 

118 

119 Args: 

120 logs_dir: Directory for session log files 

121 force: If True, re-registers even if already registered 

122 

123 """ 

124 from session_buddy.utils.logging import SessionLogger 

125 

126 if not force: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 with suppress(Exception): 

128 existing = depends.get_sync(SessionLogger) 

129 if isinstance(existing, SessionLogger): 

130 return 

131 

132 # Create SessionLogger instance with fallback to temp logs if needed 

133 try: 

134 session_logger = SessionLogger(logs_dir) 

135 except Exception: 

136 tmp_logs = Path(tempfile.gettempdir()) / "session-buddy" / "logs" 

137 tmp_logs.mkdir(parents=True, exist_ok=True) 

138 depends.set(LOGS_DIR_KEY, tmp_logs) 

139 session_logger = SessionLogger(tmp_logs) 

140 depends.set(SessionLogger, session_logger) 

141 

142 

143def _register_permissions_manager(claude_dir: Path, force: bool) -> None: 

144 """Register SessionPermissionsManager with the given Claude directory. 

145 

146 Args: 

147 claude_dir: Root Claude directory for session data 

148 force: If True, re-registers even if already registered 

149 

150 Note: 

151 Accepts Path directly instead of resolving from string keys, 

152 following ACB's type-based dependency injection pattern. 

153 

154 """ 

155 from session_buddy.core.permissions import SessionPermissionsManager 

156 

157 if not force: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 with suppress(Exception): # Catch all DI resolution errors 

159 existing = depends.get_sync(SessionPermissionsManager) 

160 if isinstance(existing, SessionPermissionsManager): 

161 return 

162 

163 # Create and register permissions manager instance 

164 permissions_manager = SessionPermissionsManager(claude_dir) 

165 depends.set(SessionPermissionsManager, permissions_manager) 

166 

167 

168def _register_lifecycle_manager(force: bool) -> None: 

169 """Register SessionLifecycleManager with the DI container. 

170 

171 Args: 

172 force: If True, re-registers even if already registered 

173 

174 """ 

175 from session_buddy.core.session_manager import SessionLifecycleManager 

176 

177 if not force: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 with suppress(Exception): # Catch all DI resolution errors 

179 existing = depends.get_sync(SessionLifecycleManager) 

180 if isinstance(existing, SessionLifecycleManager): 

181 return 

182 

183 # Create and register lifecycle manager instance 

184 lifecycle_manager = SessionLifecycleManager() 

185 depends.set(SessionLifecycleManager, lifecycle_manager) 

186 

187 

188__all__ = [ 

189 # Legacy string keys (deprecated - use SessionPaths instead) 

190 "CLAUDE_DIR_KEY", 

191 "COMMANDS_DIR_KEY", 

192 "LOGS_DIR_KEY", 

193 "SessionPaths", 

194 "configure", 

195 "depends", 

196 "get_sync_typed", 

197 "reset", 

198]