Coverage for little_loops / issue_history / parsing.py: 48%
179 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Issue history parsing and scanning functions.
3Provides functions to parse completed issue files, extract metadata
4from frontmatter and content, scan directories for issues, and
5extract file paths from issue content.
6"""
8from __future__ import annotations
10import logging
11import re
12import subprocess
13from datetime import date
14from pathlib import Path
15from typing import Any
17from little_loops.frontmatter import parse_frontmatter
18from little_loops.issue_history.models import CompletedIssue
19from little_loops.text_utils import extract_file_paths
21logger = logging.getLogger(__name__)
24def parse_completed_issue(
25 file_path: Path, *, batch_dates: dict[str, date] | None = None
26) -> CompletedIssue:
27 """Parse a completed issue file.
29 Args:
30 file_path: Path to the issue markdown file
31 batch_dates: Optional pre-fetched mapping of filename → add-date from a batch
32 git log call; when provided, skips the per-file subprocess call.
34 Returns:
35 CompletedIssue with parsed metadata
36 """
37 filename = file_path.name
38 content = file_path.read_text(encoding="utf-8")
40 # Extract from filename: P[0-5]-[TYPE]-[NNN]-description.md
41 issue_type = "UNKNOWN"
42 priority = "P5"
43 issue_id = "UNKNOWN"
45 # Match priority
46 priority_match = re.match(r"^(P\d)", filename)
47 if priority_match:
48 priority = priority_match.group(1)
50 # Match type and ID
51 type_match = re.search(r"(BUG|ENH|FEAT)-(\d+)", filename)
52 if type_match:
53 issue_type = type_match.group(1)
54 issue_id = f"{type_match.group(1)}-{type_match.group(2)}"
56 # Parse frontmatter once for discovered_by and discovered_date
57 fm = parse_frontmatter(content)
58 discovered_by = _parse_discovered_by(fm)
59 discovered_date = _parse_discovered_date(fm)
61 # Parse completion date from Resolution section or file mtime
62 completed_date = _parse_completion_date(content, file_path, batch_dates=batch_dates)
64 return CompletedIssue(
65 path=file_path,
66 issue_type=issue_type,
67 priority=priority,
68 issue_id=issue_id,
69 discovered_by=discovered_by,
70 discovered_date=discovered_date,
71 completed_date=completed_date,
72 )
75def _parse_discovered_by(fm: dict[str, Any]) -> str | None:
76 """Extract discovered_by from parsed frontmatter.
78 Args:
79 fm: Parsed frontmatter dictionary
81 Returns:
82 discovered_by value or None
83 """
84 value = fm.get("discovered_by")
85 return value if isinstance(value, str) else None
88def _batch_completion_dates(completed_dir: Path) -> dict[str, date]:
89 """Fetch git add-dates for all files in completed_dir in one git log call.
91 Args:
92 completed_dir: Path to .issues/completed/
94 Returns:
95 Mapping from filename (basename only) to the date it was first added in git.
96 """
97 try:
98 result = subprocess.run(
99 [
100 "git",
101 "log",
102 "--diff-filter=A",
103 "--name-only",
104 "--format=%x00%as",
105 "--",
106 str(completed_dir),
107 ],
108 capture_output=True,
109 text=True,
110 cwd=completed_dir.parent,
111 )
112 except OSError:
113 return {}
115 if result.returncode != 0 or not result.stdout.strip():
116 return {}
118 dates: dict[str, date] = {}
119 current_date: date | None = None
120 for line in result.stdout.splitlines():
121 if line.startswith("\x00"):
122 try:
123 current_date = date.fromisoformat(line[1:])
124 except ValueError:
125 current_date = None
126 elif line.strip() and current_date:
127 dates[Path(line.strip()).name] = current_date
128 return dates
131def _parse_completion_date(
132 content: str, file_path: Path, *, batch_dates: dict[str, date] | None = None
133) -> date | None:
134 """Extract completion date from Resolution section or file mtime.
136 Args:
137 content: File content
138 file_path: Path for git log fallback
139 batch_dates: Optional pre-fetched mapping of filename → add-date from a batch
140 git log call; when provided, skips the per-file subprocess call if the
141 file is found in the mapping.
143 Returns:
144 Completion date or None
145 """
146 # Try Resolution section: **Completed/Fixed/Closed/Date**: YYYY-MM-DD
147 match = re.search(r"\*\*(?:Completed|Fixed|Closed|Date)\*\*:\s*(\d{4}-\d{2}-\d{2})", content)
148 if match:
149 try:
150 return date.fromisoformat(match.group(1))
151 except ValueError:
152 pass
154 # Check batch map before falling back to per-file git log
155 if batch_dates is not None:
156 return batch_dates.get(file_path.name)
158 # Fallback to git log: date file was added to completed/ in git history
159 try:
160 result = subprocess.run(
161 ["git", "log", "--diff-filter=A", "--format=%as", "-1", "--", str(file_path)],
162 capture_output=True,
163 text=True,
164 cwd=file_path.parent,
165 )
166 if result.returncode == 0 and result.stdout.strip():
167 return date.fromisoformat(result.stdout.strip())
168 except (OSError, ValueError):
169 pass
170 return None
173def _parse_resolution_action(content: str) -> str:
174 """Extract resolution action category from issue content.
176 Categorizes based on Resolution section fields:
177 - "completed": Normal completion with **Action**: fix/implement
178 - "rejected": Explicitly rejected (out of scope, not valid)
179 - "invalid": Invalid reference or spec
180 - "duplicate": Duplicate of existing issue
181 - "deferred": Deferred to future work
183 Args:
184 content: Issue file content
186 Returns:
187 Resolution category string
188 """
189 # Look for Status field patterns
190 status_match = re.search(r"\*\*Status\*\*:\s*(.+?)(?:\n|$)", content)
191 if status_match:
192 status = status_match.group(1).strip().lower()
193 if "closed" in status:
194 # Check Reason field for specific category
195 reason_match = re.search(r"\*\*Reason\*\*:\s*(.+?)(?:\n|$)", content)
196 if reason_match:
197 reason = reason_match.group(1).strip().lower()
198 if "duplicate" in reason:
199 return "duplicate"
200 if "invalid" in reason:
201 return "invalid"
202 if "deferred" in reason:
203 return "deferred"
204 if "rejected" in reason or "out of scope" in reason:
205 return "rejected"
206 # Generic closed without specific reason
207 return "rejected"
209 # Check for Action field (normal completion)
210 action_match = re.search(r"\*\*Action\*\*:\s*(.+?)(?:\n|$)", content)
211 if action_match:
212 return "completed"
214 # Default to completed if no resolution section
215 return "completed"
218def _detect_processing_agent(content: str, discovered_source: str | None = None) -> str:
219 """Detect which processing agent handled an issue.
221 Detection strategy (in priority order):
222 1. Check discovered_source field for 'll-parallel' or 'll-auto'
223 2. Check content for '**Log Type**:' field
224 3. Check content for '**Tool**:' field
225 4. Default to 'manual'
227 Args:
228 content: Issue file content
229 discovered_source: Optional discovered_source frontmatter value
231 Returns:
232 Agent name: 'll-auto', 'll-parallel', or 'manual'
233 """
234 # Check discovered_source first
235 if discovered_source:
236 source_lower = discovered_source.lower()
237 if "ll-parallel" in source_lower:
238 return "ll-parallel"
239 if "ll-auto" in source_lower:
240 return "ll-auto"
242 # Check Log Type field
243 log_type_match = re.search(r"\*\*Log Type\*\*:\s*(.+?)(?:\n|$)", content)
244 if log_type_match:
245 log_type = log_type_match.group(1).strip().lower()
246 if "ll-parallel" in log_type:
247 return "ll-parallel"
248 if "ll-auto" in log_type:
249 return "ll-auto"
251 # Check Tool field
252 tool_match = re.search(r"\*\*Tool\*\*:\s*(.+?)(?:\n|$)", content)
253 if tool_match:
254 tool = tool_match.group(1).strip().lower()
255 if "ll-parallel" in tool:
256 return "ll-parallel"
257 if "ll-auto" in tool:
258 return "ll-auto"
260 # Default to manual
261 return "manual"
264def scan_completed_issues(completed_dir: Path) -> list[CompletedIssue]:
265 """Scan completed directory for issue files.
267 Args:
268 completed_dir: Path to .issues/completed/
270 Returns:
271 List of parsed CompletedIssue objects
272 """
273 issues: list[CompletedIssue] = []
275 if not completed_dir.exists():
276 return issues
278 batch_dates = _batch_completion_dates(completed_dir)
280 for file_path in sorted(completed_dir.glob("*.md")):
281 try:
282 issue = parse_completed_issue(file_path, batch_dates=batch_dates)
283 issues.append(issue)
284 except Exception as e:
285 logger.warning("Failed to parse %s: %s", file_path, e)
286 continue
288 return issues
291def _parse_discovered_date(fm: dict[str, Any]) -> date | None:
292 """Extract discovered_date from parsed frontmatter.
294 Args:
295 fm: Parsed frontmatter dictionary
297 Returns:
298 discovered_date value or None
299 """
300 value = fm.get("discovered_date")
301 if not isinstance(value, str):
302 return None
303 try:
304 return date.fromisoformat(value)
305 except ValueError:
306 return None
309def _extract_subsystem(content: str) -> str | None:
310 """Extract primary subsystem/directory from issue content.
312 Args:
313 content: Issue file content
315 Returns:
316 Directory path (e.g., "scripts/little_loops/") or None
317 """
318 # Look for file paths in Location or common patterns
319 patterns = [
320 r"\*\*File\*\*:\s*`?([^`\n]+/)[^/`\n]+`?", # **File**: path/to/file.py
321 r"`([a-zA-Z_][\w/.-]+/)[^/`]+\.py`", # `path/to/file.py`
322 ]
324 for pattern in patterns:
325 match = re.search(pattern, content)
326 if match:
327 return match.group(1)
329 return None
332def _extract_paths_from_issue(content: str) -> list[str]:
333 """Extract all file paths from issue content.
335 Delegates to :func:`~little_loops.text_utils.extract_file_paths`
336 and returns results as a sorted list for backward compatibility.
338 Args:
339 content: Issue file content
341 Returns:
342 Sorted list of file paths found in content
343 """
344 return sorted(extract_file_paths(content))
347def _find_test_file(source_path: str, project_root: Path | None = None) -> str | None:
348 """Find corresponding test file for a source file.
350 Checks common test file naming patterns:
351 - tests/test_<name>.py
352 - tests/<path>/test_<name>.py
353 - <path>/test_<name>.py
354 - <path>/<name>_test.py
355 - <path>/tests/test_<name>.py
357 Args:
358 source_path: Path to source file (e.g., "src/core/processor.py")
359 project_root: Project root for anchoring existence checks. Defaults to CWD.
361 Returns:
362 Path to test file if found, None otherwise
363 """
364 if not source_path.endswith(".py"):
365 return None # Only check Python files for now
367 path = Path(source_path)
368 stem = path.stem # filename without extension
369 parent = str(path.parent) if path.parent != Path(".") else ""
371 # Generate candidate test file paths
372 candidates: list[str] = [
373 f"tests/test_{stem}.py",
374 f"{parent}/test_{stem}.py" if parent else f"test_{stem}.py",
375 f"{parent}/{stem}_test.py" if parent else f"{stem}_test.py",
376 f"{parent}/tests/test_{stem}.py" if parent else f"tests/test_{stem}.py",
377 ]
379 # Add path-aware test locations
380 if parent:
381 candidates.append(f"tests/{parent}/test_{stem}.py")
383 # Project-specific pattern for little-loops
384 # e.g., scripts/little_loops/foo.py -> scripts/tests/test_foo.py
385 if source_path.startswith("scripts/little_loops/"):
386 candidates.append(f"scripts/tests/test_{stem}.py")
388 for candidate in candidates:
389 if (project_root / candidate).exists() if project_root else Path(candidate).exists():
390 return candidate
392 return None
395def scan_active_issues(
396 issues_dir: Path,
397 category_dirs: list[str] | None = None,
398) -> list[tuple[Path, str, str, date | None]]:
399 """Scan active issue directories.
401 Args:
402 issues_dir: Path to .issues/ directory
403 category_dirs: List of category subdirectory names to scan. When
404 omitted, defaults to ``["bugs", "features", "enhancements"]`` for
405 backward compatibility. Pass ``config.issue_categories`` to
406 include custom project categories.
408 Returns:
409 List of (path, issue_type, priority, discovered_date) tuples
410 """
411 results: list[tuple[Path, str, str, date | None]] = []
413 for category_dir in category_dirs or ["bugs", "features", "enhancements"]:
414 category_path = issues_dir / category_dir
415 if not category_path.exists():
416 continue
418 for file_path in category_path.glob("*.md"):
419 filename = file_path.name
421 # Extract priority
422 priority = "P5"
423 priority_match = re.match(r"^(P\d)", filename)
424 if priority_match:
425 priority = priority_match.group(1)
427 # Extract type
428 issue_type = "UNKNOWN"
429 type_match = re.search(r"(BUG|ENH|FEAT)", filename)
430 if type_match:
431 issue_type = type_match.group(1)
433 # Extract discovered date from content
434 discovered_date = None
435 try:
436 content = file_path.read_text(encoding="utf-8")
437 fm = parse_frontmatter(content)
438 discovered_date = _parse_discovered_date(fm)
439 except Exception as e:
440 logger.warning("Failed to parse %s: %s", file_path, e)
442 results.append((file_path, issue_type, priority, discovered_date))
444 return results