Coverage for little_loops / issue_history / parsing.py: 0%
231 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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, datetime
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|EPIC)-(\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, discovered_date, captured_at
57 fm = parse_frontmatter(content)
58 discovered_by = _parse_discovered_by(fm)
59 captured_at = _parse_captured_at(fm)
60 discovered_date = _parse_discovered_date(fm)
62 # Parse completion date from Resolution section or file mtime
63 completed_at = _parse_completed_at(fm)
64 completed_date = _parse_completion_date(content, file_path, batch_dates=batch_dates, fm=fm)
66 return CompletedIssue(
67 path=file_path,
68 issue_type=issue_type,
69 priority=priority,
70 issue_id=issue_id,
71 discovered_by=discovered_by,
72 discovered_date=discovered_date,
73 completed_date=completed_date,
74 captured_at=captured_at,
75 completed_at=completed_at,
76 )
79def _parse_iso_datetime(value: Any) -> datetime | None:
80 """Parse an ISO 8601 string into a naive datetime, or return None.
82 Strips a trailing ``Z`` for Python <3.11 compatibility (same convention as
83 the sibling ``cli/issues/search.py`` implementation).
84 """
85 if not isinstance(value, str) or not value:
86 return None
87 try:
88 return datetime.fromisoformat(value.rstrip("Z")).replace(tzinfo=None)
89 except ValueError:
90 return None
93def _parse_captured_at(fm: dict[str, Any]) -> datetime | None:
94 """Extract captured_at datetime from parsed frontmatter."""
95 return _parse_iso_datetime(fm.get("captured_at"))
98def _parse_completed_at(fm: dict[str, Any]) -> datetime | None:
99 """Extract completed_at datetime from parsed frontmatter."""
100 return _parse_iso_datetime(fm.get("completed_at"))
103def _parse_discovered_by(fm: dict[str, Any]) -> str | None:
104 """Extract discovered_by from parsed frontmatter.
106 Args:
107 fm: Parsed frontmatter dictionary
109 Returns:
110 discovered_by value or None
111 """
112 value = fm.get("discovered_by")
113 return value if isinstance(value, str) else None
116def _batch_completion_dates(_issues_dir: Path) -> dict[str, date]:
117 """No-op stub kept for legacy callers.
119 The previous implementation used ``git log --diff-filter=A`` against
120 ``completed/`` to detect when issue files were *moved* into completion.
121 With status decoupled from directory location (ENH-1418), files no
122 longer move on completion; ``completed_at:`` frontmatter is the primary
123 source of truth, with a per-file ``git log -1`` fallback in
124 ``_parse_completion_date``. ENH-1420 will backfill ``completed_at`` for
125 pre-decoupling issues, after which the per-file fallback can also be
126 removed.
128 Args:
129 _issues_dir: Unused (kept for signature stability).
131 Returns:
132 An empty mapping.
133 """
134 return {}
137def _parse_completion_date(
138 content: str,
139 file_path: Path,
140 *,
141 batch_dates: dict[str, date] | None = None,
142 fm: dict[str, Any] | None = None,
143) -> date | None:
144 """Extract completion date from frontmatter, Resolution section, or git log.
146 Checks ``completed_at`` frontmatter first (coerced to ``date`` via ``.date()``
147 to preserve the existing return type); then the Resolution section regex;
148 then falls back to batch_dates or a per-file git log call.
150 Args:
151 content: File content
152 file_path: Path for git log fallback
153 batch_dates: Optional pre-fetched mapping of filename → add-date from a batch
154 git log call; when provided, skips the per-file subprocess call if the
155 file is found in the mapping.
156 fm: Optional pre-parsed frontmatter dict. When absent, frontmatter is
157 parsed from ``content`` so external callers with no ``fm`` benefit
158 from the ``completed_at`` check transparently.
160 Returns:
161 Completion date or None
162 """
163 # Try completed_at frontmatter first (sub-day resolution source of truth)
164 if fm is None:
165 fm = parse_frontmatter(content)
166 completed_at = _parse_completed_at(fm)
167 if completed_at is not None:
168 return completed_at.date()
170 # Try Resolution section: **Completed/Fixed/Closed/Date**: YYYY-MM-DD
171 match = re.search(r"\*\*(?:Completed|Fixed|Closed|Date)\*\*:\s*(\d{4}-\d{2}-\d{2})", content)
172 if match:
173 try:
174 return date.fromisoformat(match.group(1))
175 except ValueError:
176 pass
178 # Check batch map before falling back to per-file git log
179 if batch_dates is not None:
180 return batch_dates.get(file_path.name)
182 # Fallback to git log: most recent commit date for this file (typically
183 # the close/done commit, since status writes are the latest change).
184 try:
185 result = subprocess.run(
186 ["git", "log", "--format=%as", "-1", "--", str(file_path)],
187 capture_output=True,
188 text=True,
189 cwd=file_path.parent,
190 )
191 if result.returncode == 0 and result.stdout.strip():
192 return date.fromisoformat(result.stdout.strip())
193 except (OSError, ValueError):
194 pass
195 return None
198def _parse_resolution_action(content: str) -> str:
199 """Extract resolution action category from issue content.
201 Categorizes based on Resolution section fields:
202 - "completed": Normal completion with **Action**: fix/implement
203 - "rejected": Explicitly rejected (out of scope, not valid)
204 - "invalid": Invalid reference or spec
205 - "duplicate": Duplicate of existing issue
206 - "deferred": Deferred to future work
208 Args:
209 content: Issue file content
211 Returns:
212 Resolution category string
213 """
214 # Look for Status field patterns
215 status_match = re.search(r"\*\*Status\*\*:\s*(.+?)(?:\n|$)", content)
216 if status_match:
217 status = status_match.group(1).strip().lower()
218 if "closed" in status:
219 # Check Reason field for specific category
220 reason_match = re.search(r"\*\*Reason\*\*:\s*(.+?)(?:\n|$)", content)
221 if reason_match:
222 reason = reason_match.group(1).strip().lower()
223 if "duplicate" in reason:
224 return "duplicate"
225 if "invalid" in reason:
226 return "invalid"
227 if "deferred" in reason:
228 return "deferred"
229 if "rejected" in reason or "out of scope" in reason:
230 return "rejected"
231 # Generic closed without specific reason
232 return "rejected"
234 # Check for Action field (normal completion)
235 action_match = re.search(r"\*\*Action\*\*:\s*(.+?)(?:\n|$)", content)
236 if action_match:
237 return "completed"
239 # Default to completed if no resolution section
240 return "completed"
243def _detect_processing_agent(content: str, discovered_source: str | None = None) -> str:
244 """Detect which processing agent handled an issue.
246 Detection strategy (in priority order):
247 1. Check discovered_source field for 'll-parallel' or 'll-auto'
248 2. Check content for '**Log Type**:' field
249 3. Check content for '**Tool**:' field
250 4. Default to 'manual'
252 Args:
253 content: Issue file content
254 discovered_source: Optional discovered_source frontmatter value
256 Returns:
257 Agent name: 'll-auto', 'll-parallel', or 'manual'
258 """
259 # Check discovered_source first
260 if discovered_source:
261 source_lower = discovered_source.lower()
262 if "ll-parallel" in source_lower:
263 return "ll-parallel"
264 if "ll-auto" in source_lower:
265 return "ll-auto"
267 # Check Log Type field
268 log_type_match = re.search(r"\*\*Log Type\*\*:\s*(.+?)(?:\n|$)", content)
269 if log_type_match:
270 log_type = log_type_match.group(1).strip().lower()
271 if "ll-parallel" in log_type:
272 return "ll-parallel"
273 if "ll-auto" in log_type:
274 return "ll-auto"
276 # Check Tool field
277 tool_match = re.search(r"\*\*Tool\*\*:\s*(.+?)(?:\n|$)", content)
278 if tool_match:
279 tool = tool_match.group(1).strip().lower()
280 if "ll-parallel" in tool:
281 return "ll-parallel"
282 if "ll-auto" in tool:
283 return "ll-auto"
285 # Default to manual
286 return "manual"
289def scan_completed_issues(
290 issues_dir: Path,
291 category_dirs: list[str] | None = None,
292) -> list[CompletedIssue]:
293 """Scan type directories for issues with ``status: done`` frontmatter.
295 Files no longer move into a ``completed/`` subdirectory on completion
296 (ENH-1418). Completion is detected by ``status: done`` in the file's
297 YAML frontmatter; files remain in their original type directory
298 (``bugs/``, ``features/``, ``enhancements/``, ``epics/``).
300 For backwards compatibility with pre-decoupling repos, a sibling
301 ``completed/`` directory under ``issues_dir`` is also scanned when
302 present so legacy completed issues continue to surface.
304 Args:
305 issues_dir: Path to ``.issues/`` (the parent of category dirs).
306 category_dirs: Optional override of category subdirectories to scan.
307 Defaults to ``["bugs", "features", "enhancements", "epics"]``.
309 Returns:
310 List of parsed ``CompletedIssue`` objects, sorted by file path.
311 """
312 issues: list[CompletedIssue] = []
314 if not issues_dir.exists():
315 return issues
317 scan_dirs = category_dirs or ["bugs", "features", "enhancements", "epics"]
318 paths_to_scan: list[Path] = []
319 for category_dir in scan_dirs:
320 category_path = issues_dir / category_dir
321 if not category_path.exists():
322 continue
323 for file_path in category_path.glob("*.md"):
324 try:
325 content = file_path.read_text(encoding="utf-8")
326 fm = parse_frontmatter(content)
327 except Exception as e:
328 logger.warning("Failed to read %s: %s", file_path, e)
329 continue
330 if fm.get("status") != "done":
331 continue
332 paths_to_scan.append(file_path)
334 # Legacy completed/ directory (pre-ENH-1418); scan unconditionally
335 # so older repos keep working until ENH-1420 backfills.
336 legacy_completed = issues_dir / "completed"
337 if legacy_completed.exists():
338 paths_to_scan.extend(legacy_completed.glob("*.md"))
340 for file_path in sorted(paths_to_scan):
341 try:
342 issue = parse_completed_issue(file_path)
343 issues.append(issue)
344 except Exception as e:
345 logger.warning("Failed to parse %s: %s", file_path, e)
346 continue
348 return issues
351def scan_completed_issues_from_db(db_path: Path) -> list[CompletedIssue]:
352 """Read completed-issue summary rows from the unified session DB (ENH-1621).
354 Queries the v2 ``issue_events`` table for rows with ``transition='done'``
355 and rebuilds :class:`CompletedIssue` dataclasses. Only the summary-relevant
356 fields (path, type, priority, id, completion timestamps) are populated —
357 ``analyze`` / ``export`` paths that need file bodies or git history must
358 continue to use :func:`scan_completed_issues`.
360 Returns ``[]`` when the DB is missing or the table has no done rows; the
361 caller treats an empty result as a signal to fall back to file parsing.
362 """
363 from little_loops.session_store import connect
365 if not db_path.exists():
366 return []
367 try:
368 conn = connect(db_path)
369 except Exception as exc: # pragma: no cover - defensive
370 logger.warning("Failed to open session DB %s: %s", db_path, exc)
371 return []
373 issues: list[CompletedIssue] = []
374 try:
375 try:
376 rows = conn.execute(
377 "SELECT issue_id, issue_type, priority, discovered_by, "
378 "captured_at, completed_at, completed_date "
379 "FROM issue_events WHERE transition = 'done'"
380 ).fetchall()
381 except Exception as exc: # pragma: no cover - schema mismatch / corruption
382 logger.warning("issue_events read failed for %s: %s", db_path, exc)
383 return []
384 finally:
385 conn.close()
387 for row in rows:
388 issue_id = row["issue_id"] or "UNKNOWN"
389 issue_type = row["issue_type"] or "UNKNOWN"
390 priority = row["priority"] or "P5"
391 captured_at = _parse_iso_datetime(row["captured_at"])
392 completed_at = _parse_iso_datetime(row["completed_at"])
393 completed_date_val = row["completed_date"]
394 completed_date: date | None = None
395 if isinstance(completed_date_val, str) and completed_date_val:
396 try:
397 completed_date = date.fromisoformat(completed_date_val[:10])
398 except ValueError:
399 completed_date = None
400 if completed_date is None and completed_at is not None:
401 completed_date = completed_at.date()
402 # discovered_date is not stored as a discrete column; derive from
403 # captured_at when present (mirrors `_parse_discovered_date`).
404 discovered_date = captured_at.date() if captured_at is not None else None
405 issues.append(
406 CompletedIssue(
407 path=Path(""), # not tracked in the DB row
408 issue_type=issue_type,
409 priority=priority,
410 issue_id=issue_id,
411 discovered_by=row["discovered_by"],
412 discovered_date=discovered_date,
413 completed_date=completed_date,
414 captured_at=captured_at,
415 completed_at=completed_at,
416 )
417 )
418 return issues
421def _parse_discovered_date(fm: dict[str, Any]) -> date | None:
422 """Extract discovered date from parsed frontmatter.
424 Prefers ``captured_at`` (ISO datetime, sub-day resolution) when present,
425 coercing via ``.date()`` to preserve the legacy ``date | None`` return type
426 so callers in ``summary.py`` / ``analysis.py`` / ``cli/history.py`` don't
427 need ``.date()`` adjustments. Falls back to ``discovered_date`` on absence
428 or parse failure.
430 Args:
431 fm: Parsed frontmatter dictionary
433 Returns:
434 Discovered date or None
435 """
436 captured = _parse_captured_at(fm)
437 if captured is not None:
438 return captured.date()
440 value = fm.get("discovered_date")
441 if not isinstance(value, str):
442 return None
443 try:
444 return date.fromisoformat(value)
445 except ValueError:
446 return None
449def _extract_subsystem(content: str) -> str | None:
450 """Extract primary subsystem/directory from issue content.
452 Args:
453 content: Issue file content
455 Returns:
456 Directory path (e.g., "scripts/little_loops/") or None
457 """
458 # Look for file paths in Location or common patterns
459 patterns = [
460 r"\*\*File\*\*:\s*`?([^`\n]+/)[^/`\n]+`?", # **File**: path/to/file.py
461 r"`([a-zA-Z_][\w/.-]+/)[^/`]+\.py`", # `path/to/file.py`
462 ]
464 for pattern in patterns:
465 match = re.search(pattern, content)
466 if match:
467 return match.group(1)
469 return None
472def _extract_paths_from_issue(content: str) -> list[str]:
473 """Extract all file paths from issue content.
475 Delegates to :func:`~little_loops.text_utils.extract_file_paths`
476 and returns results as a sorted list for backward compatibility.
478 Args:
479 content: Issue file content
481 Returns:
482 Sorted list of file paths found in content
483 """
484 return sorted(extract_file_paths(content))
487def _find_test_file(source_path: str, project_root: Path | None = None) -> str | None:
488 """Find corresponding test file for a source file.
490 Checks common test file naming patterns:
491 - tests/test_<name>.py
492 - tests/<path>/test_<name>.py
493 - <path>/test_<name>.py
494 - <path>/<name>_test.py
495 - <path>/tests/test_<name>.py
497 Args:
498 source_path: Path to source file (e.g., "src/core/processor.py")
499 project_root: Project root for anchoring existence checks. Defaults to CWD.
501 Returns:
502 Path to test file if found, None otherwise
503 """
504 if not source_path.endswith(".py"):
505 return None # Only check Python files for now
507 path = Path(source_path)
508 stem = path.stem # filename without extension
509 parent = str(path.parent) if path.parent != Path(".") else ""
511 # Generate candidate test file paths
512 candidates: list[str] = [
513 f"tests/test_{stem}.py",
514 f"{parent}/test_{stem}.py" if parent else f"test_{stem}.py",
515 f"{parent}/{stem}_test.py" if parent else f"{stem}_test.py",
516 f"{parent}/tests/test_{stem}.py" if parent else f"tests/test_{stem}.py",
517 ]
519 # Add path-aware test locations
520 if parent:
521 candidates.append(f"tests/{parent}/test_{stem}.py")
523 # Project-specific pattern for little-loops
524 # e.g., scripts/little_loops/foo.py -> scripts/tests/test_foo.py
525 if source_path.startswith("scripts/little_loops/"):
526 candidates.append(f"scripts/tests/test_{stem}.py")
528 for candidate in candidates:
529 if (project_root / candidate).exists() if project_root else Path(candidate).exists():
530 return candidate
532 return None
535def scan_active_issues(
536 issues_dir: Path,
537 category_dirs: list[str] | None = None,
538) -> list[tuple[Path, str, str, date | None]]:
539 """Scan active issue directories.
541 Args:
542 issues_dir: Path to .issues/ directory
543 category_dirs: List of category subdirectory names to scan. When
544 omitted, defaults to ``["bugs", "features", "enhancements"]`` for
545 backward compatibility. Pass ``config.issue_categories`` to
546 include custom project categories.
548 Returns:
549 List of (path, issue_type, priority, discovered_date) tuples
550 """
551 results: list[tuple[Path, str, str, date | None]] = []
553 for category_dir in category_dirs or ["bugs", "features", "enhancements"]:
554 category_path = issues_dir / category_dir
555 if not category_path.exists():
556 continue
558 for file_path in category_path.glob("*.md"):
559 filename = file_path.name
561 # Extract priority
562 priority = "P5"
563 priority_match = re.match(r"^(P\d)", filename)
564 if priority_match:
565 priority = priority_match.group(1)
567 # Extract type
568 issue_type = "UNKNOWN"
569 type_match = re.search(r"(BUG|ENH|FEAT|EPIC)", filename)
570 if type_match:
571 issue_type = type_match.group(1)
573 # Extract discovered date from content
574 discovered_date = None
575 try:
576 content = file_path.read_text(encoding="utf-8")
577 fm = parse_frontmatter(content)
578 discovered_date = _parse_discovered_date(fm)
579 except Exception as e:
580 logger.warning("Failed to parse %s: %s", file_path, e)
582 results.append((file_path, issue_type, priority, discovered_date))
584 return results