Coverage for little_loops / issue_discovery / search.py: 0%
187 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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 Status is read from each file's YAML ``status:`` frontmatter (ENH-1418).
42 Files live in their type directories (``bugs/``, ``features/`` etc.)
43 regardless of completion state. ``is_completed`` in the returned tuples
44 is ``True`` for done/cancelled/deferred issues (i.e. non-active).
46 For backwards compatibility, files in legacy ``completed/`` and
47 ``deferred/`` sibling directories are also surfaced when their
48 respective ``include_*`` flag is set.
50 Args:
51 config: Project configuration
52 include_completed: Whether to include completed/cancelled issues
53 include_deferred: Whether to include deferred issues
55 Returns:
56 List of ``(path, is_completed)`` tuples.
57 """
58 from little_loops.frontmatter import parse_frontmatter
60 files: list[tuple[Path, bool]] = []
62 for category in config.issue_categories:
63 issue_dir = config.get_issue_dir(category)
64 if not issue_dir.exists():
65 continue
66 for f in issue_dir.glob("*.md"):
67 try:
68 fm = parse_frontmatter(f.read_text(encoding="utf-8"))
69 except Exception:
70 files.append((f, False))
71 continue
72 status = fm.get("status", "open")
73 if status in ("done", "cancelled"):
74 if include_completed:
75 files.append((f, True))
76 elif status == "deferred":
77 if include_deferred:
78 files.append((f, True))
79 else:
80 files.append((f, False))
82 # Legacy completed/ and deferred/ sibling dirs (pre-ENH-1418)
83 if include_completed:
84 legacy_completed = config.get_completed_dir()
85 if legacy_completed.exists():
86 for f in legacy_completed.glob("*.md"):
87 files.append((f, True))
89 if include_deferred:
90 legacy_deferred = config.get_deferred_dir()
91 if legacy_deferred.exists():
92 for f in legacy_deferred.glob("*.md"):
93 files.append((f, True))
95 return files
98def search_issues_by_content(
99 config: BRConfig,
100 search_terms: list[str],
101 include_completed: bool = True,
102) -> list[tuple[Path, float, bool]]:
103 """Search issues by content with relevance scoring.
105 Args:
106 config: Project configuration
107 search_terms: Terms to search for
108 include_completed: Whether to include completed issues
110 Returns:
111 List of (path, score, is_completed) sorted by score descending
112 """
113 results: list[tuple[Path, float, bool]] = []
114 search_words = set()
115 for term in search_terms:
116 search_words.update(_extract_words(term))
118 if not search_words:
119 return results
121 for issue_path, is_completed in _get_all_issue_files(config, include_completed):
122 try:
123 content = issue_path.read_text(encoding="utf-8")
124 content_words = _extract_words(content)
125 score = _calculate_word_overlap(search_words, content_words)
126 if score > 0.1: # Minimum threshold
127 results.append((issue_path, score, is_completed))
128 except Exception:
129 continue
131 results.sort(key=lambda x: x[1], reverse=True)
132 return results
135def search_issues_by_file_path(
136 config: BRConfig,
137 file_path: str,
138 include_completed: bool = True,
139) -> list[tuple[Path, bool]]:
140 """Search for issues mentioning a specific file path.
142 Args:
143 config: Project configuration
144 file_path: File path to search for
145 include_completed: Whether to include completed issues
147 Returns:
148 List of (issue_path, is_completed) tuples
149 """
150 results: list[tuple[Path, bool]] = []
151 normalized_path = file_path.strip().lower()
153 # Also match partial paths (e.g., "module.py" matches "src/module.py")
154 path_parts = normalized_path.split("/")
155 filename = path_parts[-1] if path_parts else normalized_path
157 for issue_path, is_completed in _get_all_issue_files(config, include_completed):
158 try:
159 content = issue_path.read_text(encoding="utf-8").lower()
160 # Check for exact path or filename match
161 if normalized_path in content or filename in content:
162 results.append((issue_path, is_completed))
163 except Exception:
164 continue
166 return results
169# =============================================================================
170# Main Discovery Functions
171# =============================================================================
174def find_existing_issue(
175 config: BRConfig,
176 finding_type: str,
177 file_path: str | None,
178 finding_title: str,
179 finding_content: str,
180) -> FindingMatch:
181 """Search for an existing issue matching this finding.
183 Uses a multi-pass approach:
184 1. Exact file path match in Location sections
185 2. Title word overlap (> dup_overlap_threshold (default 0.7) = likely duplicate)
186 3. Content overlap analysis
188 For matches to completed issues, performs regression analysis to determine
189 if the match is a regression (fix broke) or invalid fix (never worked).
191 Args:
192 config: Project configuration
193 finding_type: Issue type ("BUG", "ENH", "FEAT")
194 file_path: File path from finding (if any)
195 finding_title: Title of the finding
196 finding_content: Full content/description of finding
198 Returns:
199 FindingMatch with best match details, including classification and
200 regression evidence for completed issue matches
201 """
202 exact_threshold = config.issues.duplicate_detection.exact_threshold
203 similar_threshold = config.issues.duplicate_detection.similar_threshold
205 best_match = FindingMatch(
206 issue_path=None,
207 match_type="none",
208 match_score=0.0,
209 exact_threshold=exact_threshold,
210 similar_threshold=similar_threshold,
211 )
213 # Pass 1: Exact file path match
214 if file_path:
215 path_matches = search_issues_by_file_path(config, file_path)
216 for issue_path, is_completed in path_matches:
217 try:
218 # Check if same type of finding (uses configured categories)
219 issue_type_match = _matches_issue_type(
220 finding_type, issue_path, config, is_completed
221 )
222 if issue_type_match:
223 # Determine classification
224 if is_completed:
225 classification, evidence = detect_regression_or_duplicate(
226 config, issue_path
227 )
228 else:
229 classification = MatchClassification.DUPLICATE
230 evidence = None
232 # High confidence if same file + same type
233 return FindingMatch(
234 issue_path=issue_path,
235 match_type="exact",
236 match_score=0.85,
237 is_completed=is_completed,
238 matched_terms=[file_path],
239 classification=classification,
240 regression_evidence=evidence,
241 exact_threshold=exact_threshold,
242 similar_threshold=similar_threshold,
243 )
244 except Exception:
245 continue
247 # Pass 2: Title similarity
248 title_words = _extract_words(finding_title)
249 if title_words:
250 best_pass2: tuple[Path, bool, float, list[str]] | None = None
251 best_pass2_score = best_match.match_score
252 for issue_path, is_completed in _get_all_issue_files(config):
253 try:
254 # Extract title from issue file
255 content = issue_path.read_text(encoding="utf-8")
256 title_match = re.search(r"^#\s+[\w-]+:\s*(.+)$", content, re.MULTILINE)
257 if title_match:
258 issue_title = title_match.group(1)
259 issue_words = _extract_words(issue_title)
260 overlap = _calculate_word_overlap(title_words, issue_words)
261 if (
262 overlap > config.history.capture_issue.dup_overlap_threshold
263 and overlap > best_pass2_score
264 ):
265 best_pass2_score = overlap
266 best_pass2 = (
267 issue_path,
268 is_completed,
269 overlap,
270 list(title_words & issue_words),
271 )
272 except Exception:
273 continue
275 # Determine classification once for the single best Pass 2 match
276 if best_pass2 is not None:
277 issue_path, is_completed, overlap, matched_terms = best_pass2
278 if is_completed:
279 classification, evidence = detect_regression_or_duplicate(config, issue_path)
280 else:
281 classification = MatchClassification.DUPLICATE
282 evidence = None
283 best_match = FindingMatch(
284 issue_path=issue_path,
285 match_type="similar",
286 match_score=overlap,
287 is_completed=is_completed,
288 matched_terms=matched_terms,
289 classification=classification,
290 regression_evidence=evidence,
291 exact_threshold=exact_threshold,
292 similar_threshold=similar_threshold,
293 )
295 # Pass 3: Content analysis
296 if best_match.match_score < similar_threshold:
297 content_matches = search_issues_by_content(
298 config,
299 [finding_title, finding_content],
300 )
301 best_pass3: tuple[Path, bool, float] | None = None
302 best_pass3_score = best_match.match_score
303 for issue_path, score, is_completed in content_matches[:5]: # Top 5
304 adjusted_score = score * 0.8 # Content matches are less precise
305 if adjusted_score > best_pass3_score:
306 best_pass3_score = adjusted_score
307 best_pass3 = (issue_path, is_completed, adjusted_score)
309 # Determine classification once for the single best Pass 3 match
310 if best_pass3 is not None:
311 issue_path, is_completed, adjusted_score = best_pass3
312 if is_completed:
313 classification, evidence = detect_regression_or_duplicate(config, issue_path)
314 else:
315 classification = MatchClassification.DUPLICATE
316 evidence = None
317 best_match = FindingMatch(
318 issue_path=issue_path,
319 match_type="content",
320 match_score=adjusted_score,
321 is_completed=is_completed,
322 classification=classification,
323 regression_evidence=evidence,
324 exact_threshold=exact_threshold,
325 similar_threshold=similar_threshold,
326 )
328 # If no match found, classification is NEW_ISSUE (the default)
329 return best_match
332# =============================================================================
333# Issue Reopening and Updating
334# =============================================================================
337def _get_category_from_issue_path(issue_path: Path, config: BRConfig) -> str:
338 """Determine the category for an issue based on its filename.
340 Args:
341 issue_path: Path to issue file
342 config: Project configuration
344 Returns:
345 Category name (e.g., "bugs", "enhancements", "features")
346 """
347 filename = issue_path.name.upper()
348 for category_name, category_config in config.issues.categories.items():
349 if category_config.prefix in filename:
350 return category_name
351 return "bugs" # Default
354def reopen_issue(
355 config: BRConfig,
356 completed_issue_path: Path,
357 reopen_reason: str,
358 new_context: str,
359 source_command: str,
360 logger: Logger,
361 classification: MatchClassification | None = None,
362 regression_evidence: RegressionEvidence | None = None,
363) -> Path | None:
364 """Move issue from completed back to active with Reopened section.
366 Args:
367 config: Project configuration
368 completed_issue_path: Path to issue in completed/
369 reopen_reason: Reason for reopening
370 new_context: New context/findings to add
371 source_command: Command triggering the reopen
372 logger: Logger for output
373 classification: How this issue was classified (regression, invalid_fix, etc.)
374 regression_evidence: Evidence supporting the classification
376 Returns:
377 New path to reopened issue, or None if failed
378 """
379 if not completed_issue_path.exists():
380 logger.error(f"Completed issue not found: {completed_issue_path}")
381 return None
383 # Determine target category directory
384 category = _get_category_from_issue_path(completed_issue_path, config)
385 target_dir = config.get_issue_dir(category)
386 target_dir.mkdir(parents=True, exist_ok=True)
388 target_path = target_dir / completed_issue_path.name
390 # Safety check - don't overwrite a *different* active issue at the
391 # target. If the completed_issue_path is already at target_path
392 # (post-ENH-1418: file lives in its type dir), this is the same file
393 # and we just rewrite frontmatter in place.
394 if target_path.exists() and target_path.resolve() != completed_issue_path.resolve():
395 logger.warning(f"Active issue already exists: {target_path}")
396 return None
398 # Log with classification info if available
399 if classification == MatchClassification.REGRESSION:
400 logger.info(f"Reopening {completed_issue_path.name} as REGRESSION -> {category}/")
401 elif classification == MatchClassification.INVALID_FIX:
402 logger.info(f"Reopening {completed_issue_path.name} as INVALID_FIX -> {category}/")
403 else:
404 logger.info(f"Reopening {completed_issue_path.name} -> {category}/")
406 try:
407 from little_loops.frontmatter import update_frontmatter
409 content = completed_issue_path.read_text(encoding="utf-8")
411 reopen_section = _build_reopen_section(
412 reopen_reason,
413 new_context,
414 source_command,
415 classification,
416 regression_evidence,
417 )
418 content += reopen_section
419 content = update_frontmatter(content, {"status": "open"})
421 same_file = completed_issue_path.resolve() == target_path.resolve()
422 if same_file:
423 # Post-ENH-1418: file already in its type dir; just rewrite content.
424 completed_issue_path.write_text(content, encoding="utf-8")
425 else:
426 # Legacy path: file lives in completed/ — move it back to the type dir.
427 result = subprocess.run(
428 ["git", "mv", str(completed_issue_path), str(target_path)],
429 capture_output=True,
430 text=True,
431 )
432 if result.returncode != 0:
433 logger.warning(f"git mv failed, using manual copy: {result.stderr}")
434 target_path.write_text(content, encoding="utf-8")
435 completed_issue_path.unlink()
436 else:
437 target_path.write_text(content, encoding="utf-8")
439 logger.success(f"Reopened: {target_path.name}")
440 return target_path
442 except Exception as e:
443 logger.error(f"Failed to reopen issue: {e}")
444 return None
447def update_existing_issue(
448 config: BRConfig,
449 issue_path: Path,
450 update_section_name: str,
451 update_content: str,
452 source_command: str,
453 logger: Logger,
454) -> bool:
455 """Add new findings to an existing issue.
457 Args:
458 config: Project configuration
459 issue_path: Path to issue file
460 update_section_name: Name for the update section
461 update_content: Content to add
462 source_command: Command triggering the update
463 logger: Logger for output
465 Returns:
466 True if update succeeded
467 """
468 if not issue_path.exists():
469 logger.error(f"Issue not found: {issue_path}")
470 return False
472 try:
473 content = issue_path.read_text(encoding="utf-8")
475 # Build update section
476 update_section = f"""
478---
480## {update_section_name}
482- **Date**: {datetime.now().strftime("%Y-%m-%d")}
483- **Source**: {source_command}
485{update_content}
486"""
488 # Check if section already exists
489 if f"## {update_section_name}" not in content:
490 content += update_section
491 issue_path.write_text(content, encoding="utf-8")
492 logger.success(f"Updated: {issue_path.name}")
493 else:
494 logger.info(f"Section already exists in {issue_path.name}, skipping")
496 return True
498 except Exception as e:
499 logger.error(f"Failed to update issue: {e}")
500 return False