Coverage for session_buddy / utils / reflection_utils.py: 61.46%

1"""Reflection storage utilities for intelligent checkpoint management. 

2 

3This module provides utilities for determining when checkpoint reflections 

4should be automatically stored, balancing signal-to-noise ratio with 

5comprehensive session memory capture. 

6""" 

7 

8import typing as t 

9from enum import Enum 

10 

11from session_buddy.settings import get_settings 

12 

13 

14class CheckpointReason(Enum): 

15 """Reasons for automatic reflection storage.""" 

16 

17 MANUAL_CHECKPOINT = "manual_checkpoint" 

18 SESSION_END = "session_end" 

19 QUALITY_IMPROVEMENT = "quality_improvement" 

20 QUALITY_DEGRADATION = "quality_degradation" 

21 EXCEPTIONAL_QUALITY = "exceptional_quality" 

22 ROUTINE_SKIP = "routine_skip" 

23 

24 

25class AutoStoreDecision(t.NamedTuple): 

26 """Decision result for auto-storing a checkpoint reflection.""" 

27 

28 should_store: bool 

29 reason: CheckpointReason 

30 metadata: dict[str, t.Any] 

31 

32 

33def should_auto_store_checkpoint( 

34 quality_score: int, 

35 previous_score: int | None = None, 

36 is_manual: bool = False, 

37 session_phase: str = "checkpoint", 

38) -> AutoStoreDecision: 

39 """Determine if checkpoint deserves automatic reflection storage. 

40 

41 This function implements selective auto-store logic to maintain high 

42 signal-to-noise ratio in the reflection database. 

43 

44 Args: 

45 quality_score: Current session quality score (0-100) 

46 previous_score: Previous checkpoint quality score, if available 

47 is_manual: Whether this is a manually-triggered checkpoint 

48 session_phase: Session lifecycle phase ('checkpoint', 'end', 'start') 

49 

50 Returns: 

51 AutoStoreDecision with storage recommendation and reasoning 

52 

53 Examples: 

54 >>> # Manual checkpoint - always store 

55 >>> decision = should_auto_store_checkpoint(75, is_manual=True) 

56 >>> decision.should_store 

57 True 

58 >>> decision.reason 

59 <CheckpointReason.MANUAL_CHECKPOINT: 'manual_checkpoint'> 

60 

61 >>> # Significant quality improvement 

62 >>> decision = should_auto_store_checkpoint(85, previous_score=70) 

63 >>> decision.should_store 

64 True 

65 >>> decision.reason 

66 <CheckpointReason.QUALITY_IMPROVEMENT: 'quality_improvement'> 

67 

68 >>> # Routine checkpoint - skip 

69 >>> decision = should_auto_store_checkpoint(75, previous_score=73) 

70 >>> decision.should_store 

71 False 

72 >>> decision.reason 

73 <CheckpointReason.ROUTINE_SKIP: 'routine_skip'> 

74 

75 """ 

76 config = get_settings() 

77 

78 # Check if auto-store is globally enabled 

79 if not config.enable_auto_store_reflections: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 return AutoStoreDecision( 

81 should_store=False, 

82 reason=CheckpointReason.ROUTINE_SKIP, 

83 metadata={"disabled": True}, 

84 ) 

85 

86 # Always store manual checkpoints 

87 if is_manual and config.auto_store_manual_checkpoints: 

88 return AutoStoreDecision( 

89 should_store=True, 

90 reason=CheckpointReason.MANUAL_CHECKPOINT, 

91 metadata={ 

92 "quality_score": quality_score, 

93 "previous_score": previous_score, 

94 }, 

95 ) 

96 

97 # Always store session end 

98 if session_phase == "end" and config.auto_store_session_end: 

99 return AutoStoreDecision( 

100 should_store=True, 

101 reason=CheckpointReason.SESSION_END, 

102 metadata={ 

103 "quality_score": quality_score, 

104 "previous_score": previous_score, 

105 }, 

106 ) 

107 

108 # Store exceptional quality sessions 

109 if quality_score >= config.auto_store_exceptional_quality_threshold: 

110 return AutoStoreDecision( 

111 should_store=True, 

112 reason=CheckpointReason.EXCEPTIONAL_QUALITY, 

113 metadata={ 

114 "quality_score": quality_score, 

115 "threshold": config.auto_store_exceptional_quality_threshold, 

116 }, 

117 ) 

118 

119 # Store significant quality changes 

120 if previous_score is not None: 

121 quality_delta = abs(quality_score - previous_score) 

122 threshold = config.auto_store_quality_delta_threshold 

123 

124 if quality_delta >= threshold: 

125 reason = ( 

126 CheckpointReason.QUALITY_IMPROVEMENT 

127 if quality_score > previous_score 

128 else CheckpointReason.QUALITY_DEGRADATION 

129 ) 

130 return AutoStoreDecision( 

131 should_store=True, 

132 reason=reason, 

133 metadata={ 

134 "quality_score": quality_score, 

135 "previous_score": previous_score, 

136 "delta": quality_delta, 

137 "threshold": threshold, 

138 }, 

139 ) 

140 

141 # Skip routine checkpoints 

142 return AutoStoreDecision( 

143 should_store=False, 

144 reason=CheckpointReason.ROUTINE_SKIP, 

145 metadata={ 

146 "quality_score": quality_score, 

147 "previous_score": previous_score, 

148 "message": "Routine checkpoint without significant changes", 

149 }, 

150 ) 

151 

152 

153def generate_auto_store_tags( 

154 reason: CheckpointReason, 

155 project: str | None = None, 

156 quality_score: int | None = None, 

157) -> list[str]: 

158 """Generate semantic tags for auto-stored checkpoint reflections. 

159 

160 Args: 

161 reason: Why this checkpoint was auto-stored 

162 project: Project name/identifier 

163 quality_score: Session quality score 

164 

165 Returns: 

166 List of semantic tags for the reflection 

167 

168 """ 

169 tags = ["checkpoint", "auto-stored", reason.value] 

170 

171 if project: 

172 tags.append(project) 

173 

174 # Add quality-based tags 

175 if quality_score is not None: 

176 if quality_score >= 90: 

177 tags.append("high-quality") 

178 elif quality_score >= 75: 

179 tags.append("good-quality") 

180 elif quality_score < 60: 

181 tags.append("needs-improvement") 

182 

183 # Add phase-specific tags 

184 if reason == CheckpointReason.SESSION_END: 

185 tags.append("session-summary") 

186 elif reason == CheckpointReason.MANUAL_CHECKPOINT: 

187 tags.append("user-initiated") 

188 elif reason in { 

189 CheckpointReason.QUALITY_IMPROVEMENT, 

190 CheckpointReason.QUALITY_DEGRADATION, 

191 }: 

192 tags.append("quality-change") 

193 

194 return tags 

195 

196 

197def format_auto_store_summary(decision: AutoStoreDecision) -> str: 

198 """Format a human-readable summary of auto-store decision. 

199 

200 Args: 

201 decision: The AutoStoreDecision to summarize 

202 

203 Returns: 

204 Formatted summary string 

205 

206 """ 

207 if not decision.should_store: 

208 return "⏭️ Routine checkpoint - reflection storage skipped (maintains high signal-to-noise ratio)" 

209 

210 reason_messages = { 

211 CheckpointReason.MANUAL_CHECKPOINT: "💾 Manual checkpoint - reflection stored automatically", 

212 CheckpointReason.SESSION_END: "📝 Session end - final reflection stored", 

213 CheckpointReason.QUALITY_IMPROVEMENT: "📈 Quality improved significantly - reflection stored", 

214 CheckpointReason.QUALITY_DEGRADATION: "📉 Quality changed significantly - reflection stored for analysis", 

215 CheckpointReason.EXCEPTIONAL_QUALITY: "⭐ Exceptional quality session - reflection stored", 

216 } 

217 

218 message = reason_messages.get( 

219 decision.reason, 

220 "💾 Checkpoint reflection stored", 

221 ) 

222 

223 # Add metadata details 

224 if "quality_score" in decision.metadata: 224 ↛ 234line 224 didn't jump to line 234 because the condition on line 224 was always true

225 message += f" (quality: {decision.metadata['quality_score']}/100" 

226 if "delta" in decision.metadata: 226 ↛ 232line 226 didn't jump to line 232 because the condition on line 226 was always true

227 delta = decision.metadata["delta"] 

228 direction = ( 

229 "+" if decision.reason == CheckpointReason.QUALITY_IMPROVEMENT else "-" 

230 ) 

231 message += f", {direction}{delta} points" 

232 message += ")" 

233 

234 return message