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

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(completed_issues, hotspot_analysis) 

127 

128 # Rejection rate analysis 

129 rejection_analysis = analyze_rejection_rates(completed_issues, contents=issue_contents) 

130 

131 # Manual pattern analysis 

132 manual_pattern_analysis = detect_manual_patterns(completed_issues, contents=issue_contents) 

133 

134 # Agent effectiveness analysis 

135 agent_effectiveness_analysis = analyze_agent_effectiveness( 

136 completed_issues, contents=issue_contents 

137 ) 

138 

139 # Complexity proxy analysis 

140 complexity_proxy_analysis = analyze_complexity_proxy( 

141 completed_issues, hotspot_analysis, contents=issue_contents 

142 ) 

143 

144 # Configuration gaps analysis (depends on manual_pattern_analysis) 

145 config_gaps_analysis = detect_config_gaps(manual_pattern_analysis, project_root) 

146 

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

148 cross_cutting_analysis = detect_cross_cutting_smells( 

149 completed_issues, hotspot_analysis, contents=issue_contents 

150 ) 

151 

152 # Technical debt metrics 

153 debt_metrics = _calculate_debt_metrics(completed_issues, active_issues) 

154 

155 # Build analysis 

156 analysis = HistoryAnalysis( 

157 generated_date=today, 

158 total_completed=len(completed_issues), 

159 total_active=len(active_issues), 

160 date_range_start=summary.earliest_date, 

161 date_range_end=summary.latest_date, 

162 summary=summary, 

163 period_metrics=period_metrics, 

164 velocity_trend=velocity_trend, 

165 bug_ratio_trend=bug_ratio_trend, 

166 subsystem_health=subsystem_health, 

167 hotspot_analysis=hotspot_analysis, 

168 coupling_analysis=coupling_analysis, 

169 regression_analysis=regression_analysis, 

170 test_gap_analysis=test_gap_analysis, 

171 rejection_analysis=rejection_analysis, 

172 manual_pattern_analysis=manual_pattern_analysis, 

173 agent_effectiveness_analysis=agent_effectiveness_analysis, 

174 complexity_proxy_analysis=complexity_proxy_analysis, 

175 config_gaps_analysis=config_gaps_analysis, 

176 cross_cutting_analysis=cross_cutting_analysis, 

177 debt_metrics=debt_metrics, 

178 ) 

179 

180 # Comparative analysis 

181 if compare_days: 

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

183 cutoff = today - timedelta(days=compare_days) 

184 prev_cutoff = cutoff - timedelta(days=compare_days) 

185 

186 current_issues = [ 

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

188 ] 

189 previous_issues = [ 

190 i 

191 for i in completed_issues 

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

193 ] 

194 

195 if current_issues: 

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

197 for i in current_issues: 

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

199 

200 analysis.current_period = PeriodMetrics( 

201 period_start=cutoff, 

202 period_end=today, 

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

204 total_completed=len(current_issues), 

205 type_counts=current_types, 

206 ) 

207 

208 if previous_issues: 

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

210 for i in previous_issues: 

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

212 

213 analysis.previous_period = PeriodMetrics( 

214 period_start=prev_cutoff, 

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

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

217 total_completed=len(previous_issues), 

218 type_counts=prev_types, 

219 ) 

220 

221 return analysis