Coverage for little_loops / issue_discovery / search.py: 0%
168 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:19 -0500
1"""Issue file search and main discovery functions."""
3from __future__ import annotations
5import re
6import subprocess
7from datetime import datetime
8from pathlib import Path
9from typing import TYPE_CHECKING
11from little_loops.issue_discovery.extraction import (
12 _build_reopen_section,
13 detect_regression_or_duplicate,
14)
15from little_loops.issue_discovery.matching import (
16 FindingMatch,
17 MatchClassification,
18 RegressionEvidence,
19 _calculate_word_overlap,
20 _extract_words,
21 _matches_issue_type,
22)
24if TYPE_CHECKING:
25 from little_loops.config import BRConfig
26 from little_loops.logger import Logger
29# =============================================================================
30# Issue Search Functions
31# =============================================================================
34def _get_all_issue_files(
35 config: BRConfig,
36 include_completed: bool = True,
37 include_deferred: bool = False,
38) -> list[tuple[Path, bool]]:
39 """Get all issue files with their completion status.
41 Args:
42 config: Project configuration
43 include_completed: Whether to include completed issues
44 include_deferred: Whether to include deferred issues
46 Returns:
47 List of (path, is_completed) tuples.
48 For deferred issues, is_completed is set to True (non-active).
49 """
50 files: list[tuple[Path, bool]] = []
52 # Active issues
53 for category in config.issue_categories:
54 issue_dir = config.get_issue_dir(category)
55 if issue_dir.exists():
56 for f in issue_dir.glob("*.md"):
57 files.append((f, False))
59 # Completed issues
60 if include_completed:
61 completed_dir = config.get_completed_dir()
62 if completed_dir.exists():
63 for f in completed_dir.glob("*.md"):
64 files.append((f, True))
66 # Deferred issues
67 if include_deferred:
68 deferred_dir = config.get_deferred_dir()
69 if deferred_dir.exists():
70 for f in deferred_dir.glob("*.md"):
71 files.append((f, True))
73 return files
76def search_issues_by_content(
77 config: BRConfig,
78 search_terms: list[str],
79 include_completed: bool = True,
80) -> list[tuple[Path, float, bool]]:
81 """Search issues by content with relevance scoring.
83 Args:
84 config: Project configuration
85 search_terms: Terms to search for
86 include_completed: Whether to include completed issues
88 Returns:
89 List of (path, score, is_completed) sorted by score descending
90 """
91 results: list[tuple[Path, float, bool]] = []
92 search_words = set()
93 for term in search_terms:
94 search_words.update(_extract_words(term))
96 if not search_words:
97 return results
99 for issue_path, is_completed in _get_all_issue_files(config, include_completed):
100 try:
101 content = issue_path.read_text(encoding="utf-8")
102 content_words = _extract_words(content)
103 score = _calculate_word_overlap(search_words, content_words)
104 if score > 0.1: # Minimum threshold
105 results.append((issue_path, score, is_completed))
106 except Exception:
107 continue
109 results.sort(key=lambda x: x[1], reverse=True)
110 return results
113def search_issues_by_file_path(
114 config: BRConfig,
115 file_path: str,
116 include_completed: bool = True,
117) -> list[tuple[Path, bool]]:
118 """Search for issues mentioning a specific file path.
120 Args:
121 config: Project configuration
122 file_path: File path to search for
123 include_completed: Whether to include completed issues
125 Returns:
126 List of (issue_path, is_completed) tuples
127 """
128 results: list[tuple[Path, bool]] = []
129 normalized_path = file_path.strip().lower()
131 # Also match partial paths (e.g., "module.py" matches "src/module.py")
132 path_parts = normalized_path.split("/")
133 filename = path_parts[-1] if path_parts else normalized_path
135 for issue_path, is_completed in _get_all_issue_files(config, include_completed):
136 try:
137 content = issue_path.read_text(encoding="utf-8").lower()
138 # Check for exact path or filename match
139 if normalized_path in content or filename in content:
140 results.append((issue_path, is_completed))
141 except Exception:
142 continue
144 return results
147# =============================================================================
148# Main Discovery Functions
149# =============================================================================
152def find_existing_issue(
153 config: BRConfig,
154 finding_type: str,
155 file_path: str | None,
156 finding_title: str,
157 finding_content: str,
158) -> FindingMatch:
159 """Search for an existing issue matching this finding.
161 Uses a multi-pass approach:
162 1. Exact file path match in Location sections
163 2. Title word overlap (>70% = likely duplicate)
164 3. Content overlap analysis
166 For matches to completed issues, performs regression analysis to determine
167 if the match is a regression (fix broke) or invalid fix (never worked).
169 Args:
170 config: Project configuration
171 finding_type: Issue type ("BUG", "ENH", "FEAT")
172 file_path: File path from finding (if any)
173 finding_title: Title of the finding
174 finding_content: Full content/description of finding
176 Returns:
177 FindingMatch with best match details, including classification and
178 regression evidence for completed issue matches
179 """
180 exact_threshold = config.issues.duplicate_detection.exact_threshold
181 similar_threshold = config.issues.duplicate_detection.similar_threshold
183 best_match = FindingMatch(
184 issue_path=None,
185 match_type="none",
186 match_score=0.0,
187 exact_threshold=exact_threshold,
188 similar_threshold=similar_threshold,
189 )
191 # Pass 1: Exact file path match
192 if file_path:
193 path_matches = search_issues_by_file_path(config, file_path)
194 for issue_path, is_completed in path_matches:
195 try:
196 # Check if same type of finding (uses configured categories)
197 issue_type_match = _matches_issue_type(
198 finding_type, issue_path, config, is_completed
199 )
200 if issue_type_match:
201 # Determine classification
202 if is_completed:
203 classification, evidence = detect_regression_or_duplicate(
204 config, issue_path
205 )
206 else:
207 classification = MatchClassification.DUPLICATE
208 evidence = None
210 # High confidence if same file + same type
211 return FindingMatch(
212 issue_path=issue_path,
213 match_type="exact",
214 match_score=0.85,
215 is_completed=is_completed,
216 matched_terms=[file_path],
217 classification=classification,
218 regression_evidence=evidence,
219 exact_threshold=exact_threshold,
220 similar_threshold=similar_threshold,
221 )
222 except Exception:
223 continue
225 # Pass 2: Title similarity
226 title_words = _extract_words(finding_title)
227 if title_words:
228 best_pass2: tuple[Path, bool, float, list[str]] | None = None
229 best_pass2_score = best_match.match_score
230 for issue_path, is_completed in _get_all_issue_files(config):
231 try:
232 # Extract title from issue file
233 content = issue_path.read_text(encoding="utf-8")
234 title_match = re.search(r"^#\s+[\w-]+:\s*(.+)$", content, re.MULTILINE)
235 if title_match:
236 issue_title = title_match.group(1)
237 issue_words = _extract_words(issue_title)
238 overlap = _calculate_word_overlap(title_words, issue_words)
239 if overlap > 0.7 and overlap > best_pass2_score:
240 best_pass2_score = overlap
241 best_pass2 = (
242 issue_path,
243 is_completed,
244 overlap,
245 list(title_words & issue_words),
246 )
247 except Exception:
248 continue
250 # Determine classification once for the single best Pass 2 match
251 if best_pass2 is not None:
252 issue_path, is_completed, overlap, matched_terms = best_pass2
253 if is_completed:
254 classification, evidence = detect_regression_or_duplicate(config, issue_path)
255 else:
256 classification = MatchClassification.DUPLICATE
257 evidence = None
258 best_match = FindingMatch(
259 issue_path=issue_path,
260 match_type="similar",
261 match_score=overlap,
262 is_completed=is_completed,
263 matched_terms=matched_terms,
264 classification=classification,
265 regression_evidence=evidence,
266 exact_threshold=exact_threshold,
267 similar_threshold=similar_threshold,
268 )
270 # Pass 3: Content analysis
271 if best_match.match_score < similar_threshold:
272 content_matches = search_issues_by_content(
273 config,
274 [finding_title, finding_content],
275 )
276 best_pass3: tuple[Path, bool, float] | None = None
277 best_pass3_score = best_match.match_score
278 for issue_path, score, is_completed in content_matches[:5]: # Top 5
279 adjusted_score = score * 0.8 # Content matches are less precise
280 if adjusted_score > best_pass3_score:
281 best_pass3_score = adjusted_score
282 best_pass3 = (issue_path, is_completed, adjusted_score)
284 # Determine classification once for the single best Pass 3 match
285 if best_pass3 is not None:
286 issue_path, is_completed, adjusted_score = best_pass3
287 if is_completed:
288 classification, evidence = detect_regression_or_duplicate(config, issue_path)
289 else:
290 classification = MatchClassification.DUPLICATE
291 evidence = None
292 best_match = FindingMatch(
293 issue_path=issue_path,
294 match_type="content",
295 match_score=adjusted_score,
296 is_completed=is_completed,
297 classification=classification,
298 regression_evidence=evidence,
299 exact_threshold=exact_threshold,
300 similar_threshold=similar_threshold,
301 )
303 # If no match found, classification is NEW_ISSUE (the default)
304 return best_match
307# =============================================================================
308# Issue Reopening and Updating
309# =============================================================================
312def _get_category_from_issue_path(issue_path: Path, config: BRConfig) -> str:
313 """Determine the category for an issue based on its filename.
315 Args:
316 issue_path: Path to issue file
317 config: Project configuration
319 Returns:
320 Category name (e.g., "bugs", "enhancements", "features")
321 """
322 filename = issue_path.name.upper()
323 for category_name, category_config in config.issues.categories.items():
324 if category_config.prefix in filename:
325 return category_name
326 return "bugs" # Default
329def reopen_issue(
330 config: BRConfig,
331 completed_issue_path: Path,
332 reopen_reason: str,
333 new_context: str,
334 source_command: str,
335 logger: Logger,
336 classification: MatchClassification | None = None,
337 regression_evidence: RegressionEvidence | None = None,
338) -> Path | None:
339 """Move issue from completed back to active with Reopened section.
341 Args:
342 config: Project configuration
343 completed_issue_path: Path to issue in completed/
344 reopen_reason: Reason for reopening
345 new_context: New context/findings to add
346 source_command: Command triggering the reopen
347 logger: Logger for output
348 classification: How this issue was classified (regression, invalid_fix, etc.)
349 regression_evidence: Evidence supporting the classification
351 Returns:
352 New path to reopened issue, or None if failed
353 """
354 if not completed_issue_path.exists():
355 logger.error(f"Completed issue not found: {completed_issue_path}")
356 return None
358 # Determine target category directory
359 category = _get_category_from_issue_path(completed_issue_path, config)
360 target_dir = config.get_issue_dir(category)
361 target_dir.mkdir(parents=True, exist_ok=True)
363 target_path = target_dir / completed_issue_path.name
365 # Safety check - don't overwrite existing active issue
366 if target_path.exists():
367 logger.warning(f"Active issue already exists: {target_path}")
368 return None
370 # Log with classification info if available
371 if classification == MatchClassification.REGRESSION:
372 logger.info(f"Reopening {completed_issue_path.name} as REGRESSION -> {category}/")
373 elif classification == MatchClassification.INVALID_FIX:
374 logger.info(f"Reopening {completed_issue_path.name} as INVALID_FIX -> {category}/")
375 else:
376 logger.info(f"Reopening {completed_issue_path.name} -> {category}/")
378 try:
379 # Read and update content
380 content = completed_issue_path.read_text(encoding="utf-8")
382 # Add reopened section with classification info
383 reopen_section = _build_reopen_section(
384 reopen_reason,
385 new_context,
386 source_command,
387 classification,
388 regression_evidence,
389 )
390 content += reopen_section
392 # Try git mv first for history preservation
393 result = subprocess.run(
394 ["git", "mv", str(completed_issue_path), str(target_path)],
395 capture_output=True,
396 text=True,
397 )
399 if result.returncode != 0:
400 # Fall back to manual copy
401 logger.warning(f"git mv failed, using manual copy: {result.stderr}")
402 target_path.write_text(content, encoding="utf-8")
403 completed_issue_path.unlink()
404 else:
405 # Write updated content
406 target_path.write_text(content, encoding="utf-8")
408 logger.success(f"Reopened: {target_path.name}")
409 return target_path
411 except Exception as e:
412 logger.error(f"Failed to reopen issue: {e}")
413 return None
416def update_existing_issue(
417 config: BRConfig,
418 issue_path: Path,
419 update_section_name: str,
420 update_content: str,
421 source_command: str,
422 logger: Logger,
423) -> bool:
424 """Add new findings to an existing issue.
426 Args:
427 config: Project configuration
428 issue_path: Path to issue file
429 update_section_name: Name for the update section
430 update_content: Content to add
431 source_command: Command triggering the update
432 logger: Logger for output
434 Returns:
435 True if update succeeded
436 """
437 if not issue_path.exists():
438 logger.error(f"Issue not found: {issue_path}")
439 return False
441 try:
442 content = issue_path.read_text(encoding="utf-8")
444 # Build update section
445 update_section = f"""
447---
449## {update_section_name}
451- **Date**: {datetime.now().strftime("%Y-%m-%d")}
452- **Source**: {source_command}
454{update_content}
455"""
457 # Check if section already exists
458 if f"## {update_section_name}" not in content:
459 content += update_section
460 issue_path.write_text(content, encoding="utf-8")
461 logger.success(f"Updated: {issue_path.name}")
462 else:
463 logger.info(f"Section already exists in {issue_path.name}, skipping")
465 return True
467 except Exception as e:
468 logger.error(f"Failed to update issue: {e}")
469 return False