Coverage for session_mgmt_mcp/core/session_manager.py: 0.00%
161 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-01 05:22 -0700
1#!/usr/bin/env python3
2"""Session lifecycle management for session-mgmt-mcp.
4This module handles session initialization, quality assessment, checkpoints,
5and cleanup operations.
6"""
8import os
9import shutil
10from datetime import datetime
11from pathlib import Path
12from typing import Any
14from session_mgmt_mcp.utils.git_operations import (
15 create_checkpoint_commit,
16 is_git_repository,
17)
18from session_mgmt_mcp.utils.logging import get_session_logger
21class SessionLifecycleManager:
22 """Manages session lifecycle operations."""
24 def __init__(self) -> None:
25 self.logger = get_session_logger()
26 self.current_project: str | None = None
28 async def calculate_quality_score(self) -> dict[str, Any]:
29 """Calculate session quality score based on multiple factors."""
30 current_dir = Path(os.environ.get("PWD", Path.cwd()))
32 # Project health indicators (40% of score)
33 project_context = await self.analyze_project_context(current_dir)
34 project_score = (
35 sum(1 for detected in project_context.values() if detected)
36 / len(project_context)
37 ) * 40
39 # Permissions health (20% of score) - placeholder for now
40 permissions_score = 10 # Basic score until permissions system is integrated
42 # Session management availability (20% of score)
43 session_score = 20 # Always available in this refactored version
45 # Tool availability (20% of score)
46 uv_available = shutil.which("uv") is not None
47 tool_score = 20 if uv_available else 10
49 total_score = int(
50 project_score + permissions_score + session_score + tool_score,
51 )
53 return {
54 "total_score": total_score,
55 "breakdown": {
56 "project_health": project_score,
57 "permissions": permissions_score,
58 "session_management": session_score,
59 "tools": tool_score,
60 },
61 "recommendations": self._generate_quality_recommendations(
62 total_score,
63 project_context,
64 uv_available,
65 ),
66 }
68 def _generate_quality_recommendations(
69 self,
70 score: int,
71 project_context: dict,
72 uv_available: bool,
73 ) -> list[str]:
74 """Generate quality improvement recommendations based on score factors."""
75 recommendations = []
77 if score < 50:
78 recommendations.append(
79 "Session needs attention - multiple areas for improvement",
80 )
82 if not project_context.get("has_pyproject_toml", False):
83 recommendations.append(
84 "Consider adding pyproject.toml for modern Python project structure",
85 )
87 if not project_context.get("has_git_repo", False):
88 recommendations.append("Initialize git repository for version control")
90 if not uv_available:
91 recommendations.append(
92 "Install UV package manager for improved dependency management",
93 )
95 if not project_context.get("has_tests", False):
96 recommendations.append("Add test suite to improve code quality")
98 if score >= 80:
99 recommendations.append("Excellent session setup! Keep up the good work.")
100 elif score >= 60:
101 recommendations.append("Good session quality with room for optimization.")
103 return recommendations[:5] # Limit to top 5 recommendations
105 async def analyze_project_context(self, project_dir: Path) -> dict[str, bool]:
106 """Analyze project directory for common indicators and patterns."""
107 indicators = {
108 "has_pyproject_toml": (project_dir / "pyproject.toml").exists(),
109 "has_setup_py": (project_dir / "setup.py").exists(),
110 "has_requirements_txt": (project_dir / "requirements.txt").exists(),
111 "has_readme": any(
112 (project_dir / name).exists()
113 for name in ["README.md", "README.rst", "README.txt", "readme.md"]
114 ),
115 "has_git_repo": is_git_repository(project_dir),
116 "has_venv": any(
117 (project_dir / name).exists()
118 for name in [".venv", "venv", ".env", "env"]
119 ),
120 "has_tests": any(
121 (project_dir / name).exists() for name in ["tests", "test", "testing"]
122 ),
123 "has_src_structure": (project_dir / "src").exists(),
124 "has_docs": any(
125 (project_dir / name).exists() for name in ["docs", "documentation"]
126 ),
127 "has_ci_cd": any(
128 (project_dir / name).exists()
129 for name in [".github", ".gitlab-ci.yml", ".travis.yml", "Jenkinsfile"]
130 ),
131 }
133 # Additional context from file patterns
134 try:
135 python_files = list(project_dir.glob("**/*.py"))
136 indicators["has_python_files"] = len(python_files) > 0
138 # Check for common Python frameworks
139 for py_file in python_files[:10]: # Sample first 10 files
140 try:
141 with open(py_file, encoding="utf-8") as f:
142 content = f.read(1000) # Read first 1000 chars
143 if "import fastapi" in content or "from fastapi" in content:
144 indicators["uses_fastapi"] = True
145 if "import django" in content or "from django" in content:
146 indicators["uses_django"] = True
147 if "import flask" in content or "from flask" in content:
148 indicators["uses_flask"] = True
149 except (UnicodeDecodeError, PermissionError):
150 continue
152 except Exception as e:
153 self.logger.warning(f"Error analyzing Python files: {e}")
155 return indicators
157 async def perform_quality_assessment(self) -> tuple[int, dict]:
158 """Perform quality assessment and return score and data."""
159 quality_data = await self.calculate_quality_score()
160 quality_score = quality_data["total_score"]
161 return quality_score, quality_data
163 def format_quality_results(
164 self,
165 quality_score: int,
166 quality_data: dict,
167 checkpoint_result: dict | None = None,
168 ) -> list[str]:
169 """Format quality assessment results for display."""
170 output = []
172 # Quality status
173 if quality_score >= 80:
174 output.append(f"✅ Session quality: EXCELLENT (Score: {quality_score}/100)")
175 elif quality_score >= 60:
176 output.append(f"✅ Session quality: GOOD (Score: {quality_score}/100)")
177 else:
178 output.append(
179 f"⚠️ Session quality: NEEDS ATTENTION (Score: {quality_score}/100)",
180 )
182 # Quality breakdown
183 output.append("\n📈 Quality breakdown:")
184 breakdown = quality_data["breakdown"]
185 output.append(f" • Project health: {breakdown['project_health']:.1f}/40")
186 output.append(f" • Permissions: {breakdown['permissions']:.1f}/20")
187 output.append(f" • Session tools: {breakdown['session_management']:.1f}/20")
188 output.append(f" • Tool availability: {breakdown['tools']:.1f}/20")
190 # Recommendations
191 recommendations = quality_data["recommendations"]
192 if recommendations:
193 output.append("\n💡 Recommendations:")
194 for rec in recommendations[:3]:
195 output.append(f" • {rec}")
197 # Session management specific results
198 if checkpoint_result:
199 strengths = checkpoint_result.get("strengths", [])
200 if strengths:
201 output.append("\n🌟 Session strengths:")
202 for strength in strengths[:3]:
203 output.append(f" • {strength}")
205 session_stats = checkpoint_result.get("session_stats", {})
206 if session_stats:
207 output.append("\n⏱️ Session progress:")
208 output.append(
209 f" • Duration: {session_stats.get('duration_minutes', 0)} minutes",
210 )
211 output.append(
212 f" • Checkpoints: {session_stats.get('total_checkpoints', 0)}",
213 )
214 output.append(
215 f" • Success rate: {session_stats.get('success_rate', 0):.1f}%",
216 )
218 return output
220 async def perform_git_checkpoint(
221 self,
222 current_dir: Path,
223 quality_score: int,
224 ) -> list[str]:
225 """Handle git operations for checkpoint commit using the new git utilities."""
226 output = []
227 output.append("\n" + "=" * 50)
228 output.append("📦 Git Checkpoint Commit")
229 output.append("=" * 50)
231 try:
232 # Use the new git utilities
233 success, result, git_output = create_checkpoint_commit(
234 current_dir,
235 self.current_project or "Unknown",
236 quality_score,
237 )
239 output.extend(git_output)
241 if success and result != "clean":
242 self.logger.info(
243 "Checkpoint commit created",
244 project=self.current_project,
245 commit_hash=result,
246 quality_score=quality_score,
247 )
249 except Exception as e:
250 output.append(f"\n⚠️ Git operations error: {e}")
251 self.logger.exception(
252 "Git checkpoint error occurred",
253 error=str(e),
254 project=self.current_project,
255 )
257 return output
259 async def initialize_session(
260 self,
261 working_directory: str | None = None,
262 ) -> dict[str, Any]:
263 """Initialize a new session with comprehensive setup."""
264 try:
265 # Set working directory
266 if working_directory:
267 os.chdir(working_directory)
269 current_dir = Path.cwd()
270 self.current_project = current_dir.name
272 # Create .claude directory structure
273 claude_dir = Path.home() / ".claude"
274 claude_dir.mkdir(exist_ok=True)
275 (claude_dir / "data").mkdir(exist_ok=True)
276 (claude_dir / "logs").mkdir(exist_ok=True)
278 # Analyze project context
279 project_context = await self.analyze_project_context(current_dir)
280 quality_score, quality_data = await self.perform_quality_assessment()
282 self.logger.info(
283 "Session initialized",
284 project=self.current_project,
285 quality_score=quality_score,
286 working_directory=str(current_dir),
287 )
289 return {
290 "success": True,
291 "project": self.current_project,
292 "working_directory": str(current_dir),
293 "quality_score": quality_score,
294 "quality_data": quality_data,
295 "project_context": project_context,
296 "claude_directory": str(claude_dir),
297 }
299 except Exception as e:
300 self.logger.exception("Session initialization failed", error=str(e))
301 return {"success": False, "error": str(e)}
303 async def checkpoint_session(self) -> dict[str, Any]:
304 """Perform a comprehensive session checkpoint."""
305 try:
306 current_dir = Path.cwd()
307 self.current_project = current_dir.name
309 # Quality assessment
310 quality_score, quality_data = await self.perform_quality_assessment()
312 # Git checkpoint
313 git_output = await self.perform_git_checkpoint(current_dir, quality_score)
315 # Format results
316 quality_output = self.format_quality_results(quality_score, quality_data)
318 self.logger.info(
319 "Session checkpoint completed",
320 project=self.current_project,
321 quality_score=quality_score,
322 )
324 return {
325 "success": True,
326 "quality_score": quality_score,
327 "quality_output": quality_output,
328 "git_output": git_output,
329 "timestamp": datetime.now().isoformat(),
330 }
332 except Exception as e:
333 self.logger.exception("Session checkpoint failed", error=str(e))
334 return {"success": False, "error": str(e)}
336 async def end_session(self) -> dict[str, Any]:
337 """End the current session with cleanup and summary."""
338 try:
339 current_dir = Path.cwd()
340 self.current_project = current_dir.name
342 # Final quality assessment
343 quality_score, quality_data = await self.perform_quality_assessment()
345 # Create session summary
346 summary = {
347 "project": self.current_project,
348 "final_quality_score": quality_score,
349 "session_end_time": datetime.now().isoformat(),
350 "working_directory": str(current_dir),
351 "recommendations": quality_data.get("recommendations", []),
352 }
354 self.logger.info(
355 "Session ended",
356 project=self.current_project,
357 final_quality_score=quality_score,
358 )
360 return {"success": True, "summary": summary}
362 except Exception as e:
363 self.logger.exception("Session end failed", error=str(e))
364 return {"success": False, "error": str(e)}
366 async def get_session_status(
367 self,
368 working_directory: str | None = None,
369 ) -> dict[str, Any]:
370 """Get current session status and health information."""
371 try:
372 current_dir = Path(working_directory) if working_directory else Path.cwd()
374 self.current_project = current_dir.name
376 # Get comprehensive status
377 project_context = await self.analyze_project_context(current_dir)
378 quality_score, quality_data = await self.perform_quality_assessment()
380 # Check system health
381 uv_available = shutil.which("uv") is not None
382 git_available = is_git_repository(current_dir)
383 claude_dir = Path.home() / ".claude"
384 claude_dir_exists = claude_dir.exists()
386 return {
387 "success": True,
388 "project": self.current_project,
389 "working_directory": str(current_dir),
390 "quality_score": quality_score,
391 "quality_breakdown": quality_data["breakdown"],
392 "recommendations": quality_data["recommendations"],
393 "project_context": project_context,
394 "system_health": {
395 "uv_available": uv_available,
396 "git_repository": git_available,
397 "claude_directory": claude_dir_exists,
398 },
399 "timestamp": datetime.now().isoformat(),
400 }
402 except Exception as e:
403 self.logger.exception("Failed to get session status", error=str(e))
404 return {"success": False, "error": str(e)}