Coverage for little_loops / issue_history / analysis.py: 88%

1"""Issue history analysis orchestrator. 

2 

3Thin facade that coordinates all analysis sub-modules and returns 

4a comprehensive HistoryAnalysis result. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import date, timedelta 

10from pathlib import Path 

11from typing import Literal 

12 

13from little_loops.issue_history.coupling import analyze_coupling 

14from little_loops.issue_history.debt import ( 

15 _calculate_debt_metrics, 

16 analyze_agent_effectiveness, 

17 analyze_complexity_proxy, 

18 detect_cross_cutting_smells, 

19) 

20from little_loops.issue_history.hotspots import analyze_hotspots 

21from little_loops.issue_history.models import ( 

22 CompletedIssue, 

23 HistoryAnalysis, 

24 PeriodMetrics, 

25) 

26from little_loops.issue_history.parsing import scan_active_issues 

27from little_loops.issue_history.quality import ( 

28 analyze_rejection_rates, 

29 analyze_test_gaps, 

30 detect_config_gaps, 

31 detect_manual_patterns, 

32) 

33from little_loops.issue_history.regressions import analyze_regression_clustering 

34from little_loops.issue_history.summary import ( 

35 _analyze_subsystems, 

36 _calculate_trend, 

37 _group_by_period, 

38 calculate_summary, 

39) 

40 

41 

42def _load_issue_contents(issues: list[CompletedIssue]) -> dict[Path, str]: 

43 """Pre-load issue file contents for pipeline efficiency. 

44 

45 Reads each issue file once and returns a mapping from path to content. 

46 Skips unreadable files silently (matching individual function behavior). 

47 

48 Args: 

49 issues: List of completed issues to load 

50 

51 Returns: 

52 Mapping of issue path to file content 

53 """ 

54 contents: dict[Path, str] = {} 

55 for issue in issues: 

56 try: 

57 contents[issue.path] = issue.path.read_text(encoding="utf-8") 

58 except Exception: 

59 pass 

60 return contents 

61 

62 

63def calculate_analysis( 

64 completed_issues: list[CompletedIssue], 

65 issues_dir: Path | None = None, 

66 period_type: Literal["weekly", "monthly", "quarterly"] = "monthly", 

67 compare_days: int | None = None, 

68 project_root: Path | None = None, 

69) -> HistoryAnalysis: 

70 """Calculate comprehensive history analysis. 

71 

72 Args: 

73 completed_issues: List of completed issues 

74 issues_dir: Path to .issues/ for active issue scanning 

75 period_type: Grouping period for trend analysis 

76 compare_days: Days for comparative analysis (e.g., 30 for 30d comparison) 

77 project_root: Project root for config gap analysis (defaults to cwd) 

78 

79 Returns: 

80 HistoryAnalysis with all metrics 

81 """ 

82 today = date.today() 

83 

84 # Pre-load issue file contents once for all analysis functions 

85 issue_contents = _load_issue_contents(completed_issues) 

86 

87 # Get base summary 

88 summary = calculate_summary(completed_issues) 

89 

90 # Scan active issues if directory provided 

91 active_issues: list[tuple[Path, str, str, date | None]] = [] 

92 if issues_dir: 

93 active_issues = scan_active_issues(issues_dir) 

94 

95 # Calculate period metrics 

96 period_metrics = _group_by_period(completed_issues, period_type) 

97 

98 # Determine velocity trend 

99 if len(period_metrics) >= 3: 

100 velocities = [float(p.total_completed) for p in period_metrics] 

101 velocity_trend = _calculate_trend(velocities) 

102 else: 

103 velocity_trend = "stable" 

104 

105 # Determine bug ratio trend 

106 if len(period_metrics) >= 3: 

107 bug_ratios = [p.bug_ratio or 0.0 for p in period_metrics] 

108 # For bug ratio, decreasing is good (keep as-is) 

109 bug_ratio_trend = _calculate_trend(bug_ratios) 

110 else: 

111 bug_ratio_trend = "stable" 

112 

113 # Subsystem health 

114 subsystem_health = _analyze_subsystems(completed_issues, contents=issue_contents) 

115 

116 # Hotspot analysis 

117 hotspot_analysis = analyze_hotspots(completed_issues, contents=issue_contents) 

118 

119 # Coupling analysis 

120 coupling_analysis = analyze_coupling(completed_issues, contents=issue_contents) 

121 

122 # Regression clustering analysis 

123 regression_analysis = analyze_regression_clustering(completed_issues, contents=issue_contents) 

124 

125 # Test gap analysis 

126 test_gap_analysis = analyze_test_gaps( 

127 completed_issues, hotspot_analysis, project_root=project_root 

128 ) 

129 

130 # Rejection rate analysis 

131 rejection_analysis = analyze_rejection_rates(completed_issues, contents=issue_contents) 

132 

133 # Manual pattern analysis 

134 manual_pattern_analysis = detect_manual_patterns(completed_issues, contents=issue_contents) 

135 

136 # Agent effectiveness analysis 

137 agent_effectiveness_analysis = analyze_agent_effectiveness( 

138 completed_issues, contents=issue_contents 

139 ) 

140 

141 # Complexity proxy analysis 

142 complexity_proxy_analysis = analyze_complexity_proxy( 

143 completed_issues, hotspot_analysis, contents=issue_contents 

144 ) 

145 

146 # Configuration gaps analysis (depends on manual_pattern_analysis) 

147 config_gaps_analysis = detect_config_gaps(manual_pattern_analysis, project_root) 

148 

149 # Cross-cutting concern analysis (depends on hotspot_analysis) 

150 cross_cutting_analysis = detect_cross_cutting_smells( 

151 completed_issues, hotspot_analysis, contents=issue_contents 

152 ) 

153 

154 # Technical debt metrics 

155 debt_metrics = _calculate_debt_metrics(completed_issues, active_issues) 

156 

157 # Build analysis 

158 analysis = HistoryAnalysis( 

159 generated_date=today, 

160 total_completed=len(completed_issues), 

161 total_active=len(active_issues), 

162 date_range_start=summary.earliest_date, 

163 date_range_end=summary.latest_date, 

164 summary=summary, 

165 period_metrics=period_metrics, 

166 velocity_trend=velocity_trend, 

167 bug_ratio_trend=bug_ratio_trend, 

168 subsystem_health=subsystem_health, 

169 hotspot_analysis=hotspot_analysis, 

170 coupling_analysis=coupling_analysis, 

171 regression_analysis=regression_analysis, 

172 test_gap_analysis=test_gap_analysis, 

173 rejection_analysis=rejection_analysis, 

174 manual_pattern_analysis=manual_pattern_analysis, 

175 agent_effectiveness_analysis=agent_effectiveness_analysis, 

176 complexity_proxy_analysis=complexity_proxy_analysis, 

177 config_gaps_analysis=config_gaps_analysis, 

178 cross_cutting_analysis=cross_cutting_analysis, 

179 debt_metrics=debt_metrics, 

180 ) 

181 

182 # Comparative analysis 

183 if compare_days: 

184 analysis.comparison_period = f"{compare_days}d" 

185 cutoff = today - timedelta(days=compare_days) 

186 prev_cutoff = cutoff - timedelta(days=compare_days) 

187 

188 current_issues = [ 

189 i for i in completed_issues if i.completed_date and i.completed_date >= cutoff 

190 ] 

191 previous_issues = [ 

192 i 

193 for i in completed_issues 

194 if i.completed_date and prev_cutoff <= i.completed_date < cutoff 

195 ] 

196 

197 if current_issues: 

198 current_types: dict[str, int] = {} 

199 for i in current_issues: 

200 current_types[i.issue_type] = current_types.get(i.issue_type, 0) + 1 

201 

202 analysis.current_period = PeriodMetrics( 

203 period_start=cutoff, 

204 period_end=today, 

205 period_label=f"Last {compare_days} days", 

206 total_completed=len(current_issues), 

207 type_counts=current_types, 

208 ) 

209 

210 if previous_issues: 

211 prev_types: dict[str, int] = {} 

212 for i in previous_issues: 

213 prev_types[i.issue_type] = prev_types.get(i.issue_type, 0) + 1 

214 

215 analysis.previous_period = PeriodMetrics( 

216 period_start=prev_cutoff, 

217 period_end=cutoff - timedelta(days=1), 

218 period_label=f"Previous {compare_days} days", 

219 total_completed=len(previous_issues), 

220 type_counts=prev_types, 

221 ) 

222 

223 return analysis