Coverage for session_buddy / core / lifecycle / session_info.py: 62.26%

1"""Session information parsing and management. 

2 

3This module provides utilities for reading, parsing, and managing session 

4information from handoff files and session summaries. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import TYPE_CHECKING 

11 

12if TYPE_CHECKING: 

13 from pathlib import Path 

14 

15 

16@dataclass(frozen=True) 

17class SessionInfo: 

18 """Immutable session information.""" 

19 

20 session_id: str = field(default="") 

21 ended_at: str = field(default="") 

22 quality_score: str = field(default="") 

23 working_directory: str = field(default="") 

24 top_recommendation: str = field(default="") 

25 

26 def is_complete(self) -> bool: 

27 """Check if session info has required fields.""" 

28 return bool(self.ended_at and self.quality_score and self.working_directory) 

29 

30 @classmethod 

31 def empty(cls) -> SessionInfo: 

32 """Create empty session info.""" 

33 return cls() 

34 

35 @classmethod 

36 def from_dict(cls, data: dict[str, str]) -> SessionInfo: 

37 """Create from dictionary with validation.""" 

38 return cls( # type: ignore[call-arg] 

39 session_id=data.get("session_id", ""), 

40 ended_at=data.get("ended_at", ""), 

41 quality_score=data.get("quality_score", ""), 

42 working_directory=data.get("working_directory", ""), 

43 top_recommendation=data.get("top_recommendation", ""), 

44 ) 

45 

46 

47def find_latest_handoff_file(working_dir: Path) -> Path | None: 

48 """Find the most recent session handoff file.""" 

49 try: 

50 handoff_dir = working_dir / ".crackerjack" / "session" / "handoff" 

51 

52 if not handoff_dir.exists(): 

53 # Check for legacy handoff files in project root 

54 legacy_files = list(working_dir.glob("session_handoff_*.md")) 

55 if legacy_files: 

56 # Return the most recent legacy file 

57 return max(legacy_files, key=lambda f: f.stat().st_mtime) 

58 return None 

59 

60 # Find all handoff files 

61 handoff_files = list(handoff_dir.glob("session_handoff_*.md")) 

62 

63 if not handoff_files: 

64 return None 

65 

66 # Return the most recent file based on timestamp in filename 

67 return max(handoff_files, key=lambda f: f.name) 

68 

69 except Exception: 

70 return None 

71 

72 

73def discover_session_files(working_dir: Path) -> list[Path]: 

74 """Find potential session files in priority order.""" 

75 candidates = [ 

76 working_dir / "session_handoff.md", 

77 working_dir / ".claude" / "session_handoff.md", 

78 working_dir / "session_summary.md", 

79 ] 

80 return [path for path in candidates if path.exists()] 

81 

82 

83async def read_file_safely(file_path: Path) -> str: 

84 """Read file content safely.""" 

85 try: 

86 with file_path.open(encoding="utf-8") as f: 

87 return f.read() 

88 except Exception: 

89 return "" 

90 

91 

92def extract_session_metadata(lines: list[str]) -> dict[str, str]: 

93 """Extract session metadata from handoff file lines.""" 

94 info = {} 

95 for line in lines: 

96 if line.startswith("**Session ended:**"): 

97 info["ended_at"] = line.split("**Session ended:**")[1].strip() 

98 elif line.startswith("**Final quality score:**"): 

99 info["quality_score"] = line.split("**Final quality score:**")[1].strip() 

100 elif line.startswith("**Working directory:**"): 

101 info["working_directory"] = line.split("**Working directory:**")[1].strip() 

102 return info 

103 

104 

105def extract_session_recommendations(lines: list[str], info: dict[str, str]) -> None: 

106 """Extract first recommendation from recommendations section.""" 

107 in_recommendations = False 

108 for line in lines: 108 ↛ exitline 108 didn't return from function 'extract_session_recommendations' because the loop on line 108 didn't complete

109 if "## Recommendations for Next Session" in line: 

110 in_recommendations = True 

111 continue 

112 if in_recommendations and line.strip().startswith("1."): 

113 info["top_recommendation"] = line.strip()[3:].strip() # Remove "1. " 

114 break 

115 if in_recommendations and line.startswith("##"): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 break # End of recommendations section 

117 

118 

119async def parse_session_file(file_path: Path) -> SessionInfo: 

120 """Parse single session file with error handling.""" 

121 try: 

122 content = await read_file_safely(file_path) 

123 if not content: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 return SessionInfo.empty() 

125 

126 lines = content.split("\n") 

127 info_dict = extract_session_metadata(lines) 

128 extract_session_recommendations(lines, info_dict) 

129 

130 return SessionInfo.from_dict(info_dict) 

131 

132 except Exception: 

133 return SessionInfo.empty() 

134 

135 

136async def read_previous_session_info(handoff_file: Path) -> dict[str, str] | None: 

137 """Read previous session information.""" 

138 try: 

139 # Use the async parsing method 

140 session_info = await parse_session_file(handoff_file) 

141 

142 if session_info.is_complete(): 

143 return { 

144 "ended_at": session_info.ended_at, 

145 "quality_score": session_info.quality_score, 

146 "working_directory": session_info.working_directory, 

147 "top_recommendation": session_info.top_recommendation, 

148 } 

149 

150 return None 

151 

152 except Exception: 

153 return None