Coverage for little_loops / issue_history / analysis.py: 0%
73 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""Issue history analysis orchestrator.
3Thin facade that coordinates all analysis sub-modules and returns
4a comprehensive HistoryAnalysis result.
5"""
7from __future__ import annotations
9from datetime import date, timedelta
10from pathlib import Path
11from typing import Literal
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.evolution import detect_recurring_feedback, detect_skill_bypass
21from little_loops.issue_history.hotspots import analyze_hotspots
22from little_loops.issue_history.models import (
23 CompletedIssue,
24 HistoryAnalysis,
25 PeriodMetrics,
26)
27from little_loops.issue_history.parsing import scan_active_issues
28from little_loops.issue_history.quality import (
29 analyze_rejection_rates,
30 analyze_test_gaps,
31 detect_config_gaps,
32 detect_manual_patterns,
33)
34from little_loops.issue_history.regressions import analyze_regression_clustering
35from little_loops.issue_history.summary import (
36 _analyze_subsystems,
37 _calculate_trend,
38 _group_by_period,
39 calculate_summary,
40)
41from little_loops.session_store import DEFAULT_DB_PATH
44def _load_issue_contents(issues: list[CompletedIssue]) -> dict[Path, str]:
45 """Pre-load issue file contents for pipeline efficiency.
47 Reads each issue file once and returns a mapping from path to content.
48 Skips unreadable files silently (matching individual function behavior).
50 Args:
51 issues: List of completed issues to load
53 Returns:
54 Mapping of issue path to file content
55 """
56 contents: dict[Path, str] = {}
57 for issue in issues:
58 try:
59 contents[issue.path] = issue.path.read_text(encoding="utf-8")
60 except Exception:
61 pass
62 return contents
65def calculate_analysis(
66 completed_issues: list[CompletedIssue],
67 issues_dir: Path | None = None,
68 period_type: Literal["weekly", "monthly", "quarterly"] = "monthly",
69 compare_days: int | None = None,
70 project_root: Path | None = None,
71 db_path: Path | None = None,
72) -> HistoryAnalysis:
73 """Calculate comprehensive history analysis.
75 Args:
76 completed_issues: List of completed issues
77 issues_dir: Path to .issues/ for active issue scanning
78 period_type: Grouping period for trend analysis
79 compare_days: Days for comparative analysis (e.g., 30 for 30d comparison)
80 project_root: Project root for config gap analysis (defaults to cwd)
82 Returns:
83 HistoryAnalysis with all metrics
84 """
85 today = date.today()
87 # Pre-load issue file contents once for all analysis functions
88 issue_contents = _load_issue_contents(completed_issues)
90 # Get base summary
91 summary = calculate_summary(completed_issues)
93 # Scan active issues if directory provided
94 active_issues: list[tuple[Path, str, str, date | None]] = []
95 if issues_dir:
96 active_issues = scan_active_issues(issues_dir)
98 # Calculate period metrics
99 period_metrics = _group_by_period(completed_issues, period_type)
101 # Determine velocity trend
102 if len(period_metrics) >= 3:
103 velocities = [float(p.total_completed) for p in period_metrics]
104 velocity_trend = _calculate_trend(velocities)
105 else:
106 velocity_trend = "stable"
108 # Determine bug ratio trend
109 if len(period_metrics) >= 3:
110 bug_ratios = [p.bug_ratio or 0.0 for p in period_metrics]
111 # For bug ratio, decreasing is good (keep as-is)
112 bug_ratio_trend = _calculate_trend(bug_ratios)
113 else:
114 bug_ratio_trend = "stable"
116 # Subsystem health
117 subsystem_health = _analyze_subsystems(completed_issues, contents=issue_contents)
119 # Hotspot analysis
120 hotspot_analysis = analyze_hotspots(completed_issues, contents=issue_contents)
122 # Coupling analysis
123 coupling_analysis = analyze_coupling(completed_issues, contents=issue_contents)
125 # Regression clustering analysis
126 regression_analysis = analyze_regression_clustering(completed_issues, contents=issue_contents)
128 # Test gap analysis
129 test_gap_analysis = analyze_test_gaps(
130 completed_issues, hotspot_analysis, project_root=project_root
131 )
133 # Rejection rate analysis
134 rejection_analysis = analyze_rejection_rates(completed_issues, contents=issue_contents)
136 # Manual pattern analysis
137 manual_pattern_analysis = detect_manual_patterns(completed_issues, contents=issue_contents)
139 # Evolution trigger analysis (ENH-1911)
140 from little_loops.config.features import EvolutionConfig
142 _resolved_db = db_path if db_path is not None else Path(DEFAULT_DB_PATH)
143 _evolution_config = EvolutionConfig()
144 recurring_feedback_analysis = detect_recurring_feedback(
145 _resolved_db, _evolution_config, project_root=project_root
146 )
147 skill_bypass_analysis = detect_skill_bypass(
148 _resolved_db, _evolution_config, project_root=project_root
149 )
151 # Agent effectiveness analysis
152 agent_effectiveness_analysis = analyze_agent_effectiveness(
153 completed_issues, contents=issue_contents
154 )
156 # Complexity proxy analysis
157 complexity_proxy_analysis = analyze_complexity_proxy(
158 completed_issues, hotspot_analysis, contents=issue_contents
159 )
161 # Configuration gaps analysis (depends on manual_pattern_analysis)
162 config_gaps_analysis = detect_config_gaps(manual_pattern_analysis, project_root)
164 # Cross-cutting concern analysis (depends on hotspot_analysis)
165 cross_cutting_analysis = detect_cross_cutting_smells(
166 completed_issues, hotspot_analysis, contents=issue_contents
167 )
169 # Technical debt metrics
170 debt_metrics = _calculate_debt_metrics(completed_issues, active_issues)
172 # Build analysis
173 analysis = HistoryAnalysis(
174 generated_date=today,
175 total_completed=len(completed_issues),
176 total_active=len(active_issues),
177 date_range_start=summary.earliest_date,
178 date_range_end=summary.latest_date,
179 summary=summary,
180 period_metrics=period_metrics,
181 velocity_trend=velocity_trend,
182 bug_ratio_trend=bug_ratio_trend,
183 subsystem_health=subsystem_health,
184 hotspot_analysis=hotspot_analysis,
185 coupling_analysis=coupling_analysis,
186 regression_analysis=regression_analysis,
187 test_gap_analysis=test_gap_analysis,
188 rejection_analysis=rejection_analysis,
189 manual_pattern_analysis=manual_pattern_analysis,
190 agent_effectiveness_analysis=agent_effectiveness_analysis,
191 complexity_proxy_analysis=complexity_proxy_analysis,
192 config_gaps_analysis=config_gaps_analysis,
193 cross_cutting_analysis=cross_cutting_analysis,
194 recurring_feedback_analysis=recurring_feedback_analysis,
195 skill_bypass_analysis=skill_bypass_analysis,
196 debt_metrics=debt_metrics,
197 )
199 # Comparative analysis
200 if compare_days:
201 analysis.comparison_period = f"{compare_days}d"
202 cutoff = today - timedelta(days=compare_days)
203 prev_cutoff = cutoff - timedelta(days=compare_days)
205 current_issues = [
206 i for i in completed_issues if i.completed_date and i.completed_date >= cutoff
207 ]
208 previous_issues = [
209 i
210 for i in completed_issues
211 if i.completed_date and prev_cutoff <= i.completed_date < cutoff
212 ]
214 if current_issues:
215 current_types: dict[str, int] = {}
216 for i in current_issues:
217 current_types[i.issue_type] = current_types.get(i.issue_type, 0) + 1
219 analysis.current_period = PeriodMetrics(
220 period_start=cutoff,
221 period_end=today,
222 period_label=f"Last {compare_days} days",
223 total_completed=len(current_issues),
224 type_counts=current_types,
225 )
227 if previous_issues:
228 prev_types: dict[str, int] = {}
229 for i in previous_issues:
230 prev_types[i.issue_type] = prev_types.get(i.issue_type, 0) + 1
232 analysis.previous_period = PeriodMetrics(
233 period_start=prev_cutoff,
234 period_end=cutoff - timedelta(days=1),
235 period_label=f"Previous {compare_days} days",
236 total_completed=len(previous_issues),
237 type_counts=prev_types,
238 )
240 return analysis