Coverage for session_buddy / utils / quality / summary.py: 17.57%

1"""Conversation summary utilities. 

2 

3This module provides helper functions for creating and processing conversation 

4summaries, extracting key topics, decisions, and next steps from reflections. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11 

12def create_empty_summary() -> dict[str, Any]: 

13 """Create empty conversation summary structure.""" 

14 return { 

15 "key_topics": [], 

16 "decisions_made": [], 

17 "next_steps": [], 

18 "problems_solved": [], 

19 "code_changes": [], 

20 } 

21 

22 

23def extract_topics_from_content(content: str) -> set[str]: 

24 """Extract topics from reflection content.""" 

25 topics = set() 

26 if "project context:" in content: 

27 context_part = content.split("project context:")[1].split(".")[0] 

28 topics.update(word.strip() for word in context_part.split(",")) 

29 return topics 

30 

31 

32def extract_decisions_from_content(content: str) -> list[str]: 

33 """Extract decisions from reflection content.""" 

34 decisions = [] 

35 if "excellent" in content: 

36 decisions.append("Maintaining productive workflow patterns") 

37 elif "attention" in content: 

38 decisions.append("Identified areas needing workflow optimization") 

39 elif "good progress" in content: 

40 decisions.append("Steady development progress confirmed") 

41 return decisions 

42 

43 

44def extract_next_steps_from_content(content: str) -> list[str]: 

45 """Extract next steps from reflection content.""" 

46 next_steps = [] 

47 if "priority:" in content: 

48 priority_part = content.split("priority:")[1].split(".")[0] 

49 if priority_part.strip(): 

50 next_steps.append(priority_part.strip()) 

51 return next_steps 

52 

53 

54async def process_recent_reflections(db: Any, summary: dict[str, Any]) -> None: 

55 """Process recent reflections to extract conversation insights.""" 

56 recent_reflections = await db.search_reflections("checkpoint", limit=5) 

57 

58 if not recent_reflections: 

59 return 

60 

61 topics = set() 

62 decisions = [] 

63 next_steps = [] 

64 

65 for reflection in recent_reflections: 

66 content = reflection["content"].lower() 

67 

68 topics.update(extract_topics_from_content(content)) 

69 decisions.extend(extract_decisions_from_content(content)) 

70 next_steps.extend(extract_next_steps_from_content(content)) 

71 

72 summary["key_topics"] = list(topics)[:5] 

73 summary["decisions_made"] = decisions[:3] 

74 summary["next_steps"] = next_steps[:3] 

75 

76 

77def ensure_summary_defaults(summary: dict[str, Any]) -> None: 

78 """Ensure summary has default values if empty.""" 

79 if not summary["key_topics"]: 

80 summary["key_topics"] = [ 

81 "session management", 

82 "workflow optimization", 

83 ] 

84 

85 if not summary["decisions_made"]: 

86 summary["decisions_made"] = ["Proceeding with current development approach"] 

87 

88 if not summary["next_steps"]: 

89 summary["next_steps"] = ["Continue with regular checkpoint monitoring"] 

90 

91 

92def get_fallback_summary() -> dict[str, Any]: 

93 """Get fallback summary when reflection processing fails.""" 

94 return { 

95 "key_topics": ["development session", "workflow management"], 

96 "decisions_made": ["Maintaining current session approach"], 

97 "next_steps": ["Continue monitoring session quality"], 

98 "problems_solved": ["Session management optimization"], 

99 "code_changes": ["Enhanced checkpoint functionality"], 

100 } 

101 

102 

103def get_error_summary(error: Exception) -> dict[str, Any]: 

104 """Get error summary when conversation analysis fails.""" 

105 return { 

106 "key_topics": ["session analysis"], 

107 "decisions_made": ["Error during analysis"], 

108 "next_steps": ["Retry conversation summary"], 

109 "problems_solved": [], 

110 "code_changes": [], 

111 "error": str(error), 

112 }