Coverage for little_loops / issue_parser.py: 61%
256 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Issue file parsing for little-loops.
3Parses issue markdown files to extract metadata like priority, ID, type, and title.
4"""
6from __future__ import annotations
8import logging
9import re
10from dataclasses import dataclass, field
11from pathlib import Path
12from typing import TYPE_CHECKING, Any
14from little_loops.cli_args import _id_matches
15from little_loops.frontmatter import parse_frontmatter
17if TYPE_CHECKING:
18 from little_loops.config import BRConfig
21logger = logging.getLogger(__name__)
23# Regex pattern for issue IDs in list items
24# Matches: "- FEAT-001", "- BUG-123", "* ENH-005", "- FEAT-001 (some note)"
25# Also handles bold markdown: "- **ENH-1000**: description"
26ISSUE_ID_PATTERN = re.compile(r"^[-*]\s+\*{0,2}([A-Z]+-\d+)", re.MULTILINE)
29_NORMALIZED_RE = re.compile(r"^P[0-5]-(BUG|FEAT|ENH)-[0-9]{3,}-[a-z0-9-]+\.md$")
30_ISSUE_TYPE_RE = re.compile(r"-(BUG|FEAT|ENH)-")
33def is_normalized(filename: str) -> bool:
34 """Check whether an issue filename conforms to naming conventions.
36 Args:
37 filename: The basename of the issue file (e.g. 'P2-BUG-010-my-issue.md').
39 Returns:
40 True if the filename matches ``^P[0-5]-(BUG|FEAT|ENH)-[0-9]{3,}-[a-z0-9-]+\\.md$``.
41 """
42 return bool(_NORMALIZED_RE.match(filename))
45def is_formatted(issue_path: Path, templates_dir: Path | None = None) -> bool:
46 """Check whether an issue file has been formatted.
48 An issue is considered formatted if either:
49 1. Its ## Session Log contains a ``/ll:format-issue`` entry, OR
50 2. It has all required sections per its type template (structural check).
52 Args:
53 issue_path: Path to the issue markdown file.
54 templates_dir: Optional override for the templates directory.
56 Returns:
57 True if the issue is formatted by either criterion, False otherwise.
58 Returns False for files whose type cannot be determined or whose template
59 cannot be loaded.
60 """
61 from little_loops.issue_template import load_issue_sections
62 from little_loops.session_log import parse_session_log
64 try:
65 content = issue_path.read_text(encoding="utf-8")
66 except Exception:
67 return False
69 # Criterion 1: /ll:format-issue appears in the session log
70 if "/ll:format-issue" in parse_session_log(content):
71 return True
73 # Criterion 2: all required sections are present as ## headings
74 type_match = _ISSUE_TYPE_RE.search(issue_path.name)
75 if not type_match:
76 return False
77 issue_type = type_match.group(1)
79 try:
80 sections_data = load_issue_sections(issue_type, templates_dir)
81 except Exception:
82 return False
84 required: set[str] = set()
85 for name, defn in sections_data.get("common_sections", {}).items():
86 if defn.get("required") is True and not defn.get("deprecated", False):
87 required.add(name)
88 for name, defn in sections_data.get("type_sections", {}).items():
89 if defn.get("level") == "required" and not defn.get("deprecated", False):
90 required.add(name)
92 if not required:
93 return True
95 headings = {m.strip() for m in re.findall(r"^##\s+(.+)$", content, re.MULTILINE)}
96 return required.issubset(headings)
99def slugify(text: str) -> str:
100 """Convert text to slug format for filenames.
102 Args:
103 text: Text to convert
105 Returns:
106 Lowercase slug with hyphens
107 """
108 text = re.sub(r"[^\w\s-]", "", text)
109 text = re.sub(r"[-\s]+", "-", text)
110 return text.strip("-").lower()
113def get_next_issue_number(config: BRConfig, category: str | None = None) -> int:
114 """Determine the next globally unique issue number.
116 Scans ALL issue directories (active and completed) to find the highest
117 existing number across ALL issue types (BUG, FEAT, ENH). Issue numbers
118 are globally unique regardless of type.
120 Args:
121 config: Project configuration
122 category: Unused, kept for backwards compatibility
124 Returns:
125 Next available issue number (globally unique across all types)
126 """
127 max_num = 0
129 # Get all known prefixes from configuration
130 all_prefixes = [cat_config.prefix for cat_config in config.issues.categories.values()]
132 # Directories to scan: ALL category directories + completed + deferred
133 dirs_to_scan = [config.get_completed_dir(), config.get_deferred_dir()]
134 for cat_name in config.issues.categories:
135 dirs_to_scan.append(config.get_issue_dir(cat_name))
137 if not all_prefixes:
138 return max_num + 1
140 # Pre-compile a single union regex to match any known prefix
141 prefix_pattern = re.compile(r"(?:" + "|".join(re.escape(p) for p in all_prefixes) + r")-(\d+)")
143 for dir_path in dirs_to_scan:
144 if not dir_path.exists():
145 continue
146 for file in dir_path.glob("*.md"):
147 match = prefix_pattern.search(file.name)
148 if match:
149 num = int(match.group(1))
150 if num > max_num:
151 max_num = num
153 return max_num + 1
156@dataclass
157class ProductImpact:
158 """Product impact assessment for an issue.
160 Attributes:
161 goal_alignment: ID of the strategic priority this supports
162 persona_impact: ID of the persona affected
163 business_value: Business value assessment (high|medium|low)
164 user_benefit: Description of how this helps the target user
165 """
167 goal_alignment: str | None = None
168 persona_impact: str | None = None
169 business_value: str | None = None # high|medium|low
170 user_benefit: str | None = None
172 def to_dict(self) -> dict[str, Any]:
173 """Convert to dictionary for JSON serialization."""
174 return {
175 "goal_alignment": self.goal_alignment,
176 "persona_impact": self.persona_impact,
177 "business_value": self.business_value,
178 "user_benefit": self.user_benefit,
179 }
181 @classmethod
182 def from_dict(cls, data: dict[str, Any] | None) -> ProductImpact | None:
183 """Create ProductImpact from dictionary.
185 Args:
186 data: Dictionary with product impact fields, or None
188 Returns:
189 ProductImpact instance or None if data is None/empty
190 """
191 if not data:
192 return None
193 return cls(
194 goal_alignment=data.get("goal_alignment"),
195 persona_impact=data.get("persona_impact"),
196 business_value=data.get("business_value"),
197 user_benefit=data.get("user_benefit"),
198 )
201@dataclass
202class IssueInfo:
203 """Parsed information from an issue file.
205 Attributes:
206 path: Path to the issue file
207 issue_type: Type of issue (e.g., "bugs", "features")
208 priority: Priority level (e.g., "P0", "P1")
209 issue_id: Issue identifier (e.g., "BUG-123")
210 title: Issue title from markdown header
211 blocked_by: List of issue IDs that block this issue
212 blocks: List of issue IDs that this issue blocks
213 discovered_by: Source command/workflow that created this issue
214 product_impact: Product impact assessment (optional)
215 effort: Effort estimate (1=low, 2=medium, 3=high), inferred from priority if absent
216 impact: Impact estimate (1=low, 2=medium, 3=high), inferred from priority if absent
217 confidence_score: Readiness score (0-100) written by /ll:confidence-check, or None
218 outcome_confidence: Outcome confidence (0-100) written by /ll:confidence-check, or None
219 testable: Whether TDD phase should be applied; False skips TDD, None treated as testable
220 session_commands: Distinct /ll:* commands found in the ## Session Log section
221 session_command_counts: Per-command occurrence counts from the ## Session Log section
222 """
224 path: Path
225 issue_type: str
226 priority: str
227 issue_id: str
228 title: str
229 blocked_by: list[str] = field(default_factory=list)
230 blocks: list[str] = field(default_factory=list)
231 discovered_by: str | None = None
232 product_impact: ProductImpact | None = None
233 effort: int | None = None
234 impact: int | None = None
235 confidence_score: int | None = None
236 outcome_confidence: int | None = None
237 testable: bool | None = None
238 session_commands: list[str] = field(default_factory=list)
239 session_command_counts: dict[str, int] = field(default_factory=dict)
241 @property
242 def priority_int(self) -> int:
243 """Convert priority to integer for comparison (lower = higher priority)."""
244 # Support P0-P5 priorities
245 match = re.match(r"^P(\d+)$", self.priority)
246 if match:
247 return int(match.group(1))
248 return 99 # Unknown priority sorts last
250 def to_dict(self) -> dict[str, Any]:
251 """Convert to dictionary for JSON serialization."""
252 return {
253 "path": str(self.path),
254 "issue_type": self.issue_type,
255 "priority": self.priority,
256 "issue_id": self.issue_id,
257 "title": self.title,
258 "blocked_by": self.blocked_by,
259 "blocks": self.blocks,
260 "discovered_by": self.discovered_by,
261 "product_impact": (self.product_impact.to_dict() if self.product_impact else None),
262 "effort": self.effort,
263 "impact": self.impact,
264 "confidence_score": self.confidence_score,
265 "outcome_confidence": self.outcome_confidence,
266 "testable": self.testable,
267 "session_commands": self.session_commands,
268 "session_command_counts": self.session_command_counts,
269 }
271 @classmethod
272 def from_dict(cls, data: dict[str, Any]) -> IssueInfo:
273 """Create IssueInfo from dictionary."""
274 return cls(
275 path=Path(data["path"]),
276 issue_type=data["issue_type"],
277 priority=data["priority"],
278 issue_id=data["issue_id"],
279 title=data["title"],
280 blocked_by=data.get("blocked_by", []),
281 blocks=data.get("blocks", []),
282 discovered_by=data.get("discovered_by"),
283 product_impact=ProductImpact.from_dict(data.get("product_impact")),
284 effort=data.get("effort"),
285 impact=data.get("impact"),
286 confidence_score=data.get("confidence_score"),
287 outcome_confidence=data.get("outcome_confidence"),
288 testable=data.get("testable"),
289 session_commands=data.get("session_commands", []),
290 session_command_counts=data.get("session_command_counts", {}),
291 )
294class IssueParser:
295 """Parses issue files based on project configuration.
297 Uses BRConfig to understand issue categories, prefixes, and priorities.
298 """
300 def __init__(self, config: BRConfig) -> None:
301 """Initialize parser with project configuration.
303 Args:
304 config: Project configuration
305 """
306 self.config = config
307 self._build_prefix_map()
309 def _build_prefix_map(self) -> None:
310 """Build mapping from issue prefixes to category names."""
311 self._prefix_to_category: dict[str, str] = {}
312 for category_name, category in self.config.issues.categories.items():
313 self._prefix_to_category[category.prefix] = category_name
315 def parse_file(self, issue_path: Path) -> IssueInfo:
316 """Parse an issue file to extract metadata.
318 Args:
319 issue_path: Path to the issue markdown file
321 Returns:
322 Parsed IssueInfo
323 """
324 filename = issue_path.name
326 # Parse priority from filename prefix (e.g., P1-BUG-123-...)
327 priority = self._parse_priority(filename)
329 # Parse issue type and ID from filename
330 issue_type, issue_id = self._parse_type_and_id(filename, issue_path)
332 # Read content once for all content-based parsing
333 content = self._read_content(issue_path)
335 # Parse frontmatter for discovered_by, product impact, effort, and impact
336 frontmatter = parse_frontmatter(content)
337 discovered_by = frontmatter.get("discovered_by")
338 product_impact = self._parse_product_impact(frontmatter)
339 effort_raw = frontmatter.get("effort")
340 impact_raw = frontmatter.get("impact")
341 effort = int(effort_raw) if effort_raw is not None and str(effort_raw).isdigit() else None
342 impact = int(impact_raw) if impact_raw is not None and str(impact_raw).isdigit() else None
343 confidence_raw = frontmatter.get("confidence_score")
344 outcome_raw = frontmatter.get("outcome_confidence")
345 confidence_score = (
346 int(confidence_raw)
347 if confidence_raw is not None and str(confidence_raw).isdigit()
348 else None
349 )
350 outcome_confidence = (
351 int(outcome_raw) if outcome_raw is not None and str(outcome_raw).isdigit() else None
352 )
353 testable_raw = frontmatter.get("testable")
354 if isinstance(testable_raw, str):
355 testable_value: bool | None = (
356 testable_raw.lower() == "true"
357 if testable_raw.lower() in ("true", "false")
358 else None
359 )
360 else:
361 testable_value = testable_raw
363 # Parse title and dependencies from file content
364 title = self._parse_title_from_content(content, issue_path)
365 blocked_by = self._parse_blocked_by(content)
366 blocks = self._parse_blocks(content)
368 # Parse session commands from ## Session Log section
369 from little_loops.session_log import count_session_commands, parse_session_log
371 session_commands = parse_session_log(content)
372 session_command_counts = count_session_commands(content)
374 return IssueInfo(
375 path=issue_path,
376 issue_type=issue_type,
377 priority=priority,
378 issue_id=issue_id,
379 title=title,
380 blocked_by=blocked_by,
381 blocks=blocks,
382 discovered_by=discovered_by,
383 product_impact=product_impact,
384 effort=effort,
385 impact=impact,
386 confidence_score=confidence_score,
387 outcome_confidence=outcome_confidence,
388 testable=testable_value,
389 session_commands=session_commands,
390 session_command_counts=session_command_counts,
391 )
393 def _parse_priority(self, filename: str) -> str:
394 """Extract priority from filename.
396 Args:
397 filename: Issue filename
399 Returns:
400 Priority string (e.g., "P1") or last priority if not found
401 """
402 for priority in self.config.issue_priorities:
403 if filename.startswith(f"{priority}-"):
404 return priority
405 # Default to lowest priority if not found
406 return self.config.issue_priorities[-1] if self.config.issue_priorities else "P3"
408 def _get_category_for_prefix(self, prefix: str) -> str:
409 """Get category name from issue prefix.
411 Args:
412 prefix: Issue prefix (e.g., "BUG", "FEAT")
414 Returns:
415 Category name (e.g., "bugs", "features"), defaults to "bugs"
416 """
417 return self._prefix_to_category.get(prefix, "bugs")
419 def _parse_type_and_id(self, filename: str, issue_path: Path) -> tuple[str, str]:
420 """Extract issue type and ID from filename.
422 Args:
423 filename: Issue filename
424 issue_path: Full path to issue file
426 Returns:
427 Tuple of (issue_type, issue_id)
428 """
429 # Try to match known prefixes (BUG, FEAT, ENH, etc.)
430 for prefix, category in self._prefix_to_category.items():
431 pattern = rf"({prefix})-(\d+)"
432 match = re.search(pattern, filename)
433 if match:
434 issue_id = f"{match.group(1)}-{match.group(2)}"
435 return category, issue_id
437 # Fall back to inferring from directory
438 parent_name = issue_path.parent.name
439 for category_name, category_config in self.config.issues.categories.items():
440 if parent_name == category_config.dir:
441 # Generate ID from filename
442 issue_id = self._generate_id_from_filename(filename, category_config.prefix)
443 return category_name, issue_id
445 # Last resort: use filename as ID
446 return "bugs", filename.replace(".md", "")
448 def _generate_id_from_filename(self, filename: str, prefix: str) -> str:
449 """Generate an issue ID from filename when not explicitly present.
451 Args:
452 filename: Issue filename
453 prefix: Issue prefix to use
455 Returns:
456 Generated issue ID
457 """
458 # Try to extract a number from the filename
459 numbers = re.findall(r"\d+", filename)
460 if numbers:
461 return f"{prefix}-{numbers[0]}"
462 # Use next sequential number instead of hash-based fallback
463 # This ensures IDs are deterministic and don't collide with existing issues
464 category = self._get_category_for_prefix(prefix)
465 next_num = get_next_issue_number(self.config, category)
466 return f"{prefix}-{next_num:03d}"
468 def _read_content(self, issue_path: Path) -> str:
469 """Read file content, returning empty string on error.
471 Args:
472 issue_path: Path to issue file
474 Returns:
475 File content or empty string on error
476 """
477 try:
478 return issue_path.read_text(encoding="utf-8")
479 except Exception as e:
480 logger.warning("Failed to read %s: %s", issue_path.name, e)
481 return ""
483 def _parse_title_from_content(self, content: str, issue_path: Path) -> str:
484 """Extract title from issue file content.
486 Args:
487 content: Pre-read file content
488 issue_path: Path to issue file (for fallback)
490 Returns:
491 Issue title or filename stem as fallback
492 """
493 if content:
494 # Look for markdown header: # ISSUE-ID: Title
495 match = re.search(r"^#\s+[\w-]+:\s*(.+)$", content, re.MULTILINE)
496 if match:
497 return match.group(1).strip()
498 # Try first header of any format
499 match = re.search(r"^#\s+(.+)$", content, re.MULTILINE)
500 if match:
501 return match.group(1).strip()
502 # Fall back to filename
503 return issue_path.stem
505 def _parse_section_items(self, content: str, section_name: str) -> list[str]:
506 """Extract issue IDs from a markdown section.
508 Finds section header (## Section Name) and extracts issue IDs
509 from list items until the next section or end of file.
510 Skips content inside code fences.
512 Args:
513 content: File content to parse
514 section_name: Section name to find (e.g., "Blocked By")
516 Returns:
517 List of issue IDs found in the section
518 """
519 if not content:
520 return []
522 # Strip code fences to avoid matching sections in examples
523 content_without_code = self._strip_code_fences(content)
525 # Match section header case-insensitively
526 section_pattern = rf"^##\s+{re.escape(section_name)}\s*$"
527 match = re.search(section_pattern, content_without_code, re.MULTILINE | re.IGNORECASE)
528 if not match:
529 return []
531 # Get content after section header until next ## header or end
532 start = match.end()
533 next_section = re.search(r"^##\s+", content_without_code[start:], re.MULTILINE)
534 if next_section:
535 section_content = content_without_code[start : start + next_section.start()]
536 else:
537 section_content = content_without_code[start:]
539 # Extract issue IDs from list items
540 issue_ids = ISSUE_ID_PATTERN.findall(section_content)
541 return issue_ids
543 def _strip_code_fences(self, content: str) -> str:
544 """Remove code fence blocks from content.
546 Replaces content between ``` markers with empty lines to preserve
547 line numbers while removing code fence content from parsing.
549 Args:
550 content: File content
552 Returns:
553 Content with code fence blocks replaced by empty lines
554 """
555 # Match code fences: ``` or ```language through closing ```
556 result = []
557 in_fence = False
558 for line in content.split("\n"):
559 if line.startswith("```"):
560 in_fence = not in_fence
561 result.append("") # Preserve line count
562 elif in_fence:
563 result.append("") # Replace fenced content with empty line
564 else:
565 result.append(line)
566 return "\n".join(result)
568 def _parse_blocked_by(self, content: str) -> list[str]:
569 """Extract issue IDs from ## Blocked By section.
571 Args:
572 content: File content to parse
574 Returns:
575 List of issue IDs that block this issue
576 """
577 return self._parse_section_items(content, "Blocked By")
579 def _parse_blocks(self, content: str) -> list[str]:
580 """Extract issue IDs from ## Blocks section.
582 Args:
583 content: File content to parse
585 Returns:
586 List of issue IDs that this issue blocks
587 """
588 return self._parse_section_items(content, "Blocks")
590 def _parse_product_impact(self, frontmatter: dict[str, Any]) -> ProductImpact | None:
591 """Extract product impact from frontmatter.
593 Args:
594 frontmatter: Dictionary of frontmatter fields
596 Returns:
597 ProductImpact instance if any product fields are present, None otherwise
598 """
599 # Check if any product fields are present
600 product_fields = ("goal_alignment", "persona_impact", "business_value", "user_benefit")
601 if not any(frontmatter.get(key) for key in product_fields):
602 return None
604 return ProductImpact(
605 goal_alignment=frontmatter.get("goal_alignment"),
606 persona_impact=frontmatter.get("persona_impact"),
607 business_value=frontmatter.get("business_value"),
608 user_benefit=frontmatter.get("user_benefit"),
609 )
612def find_issues(
613 config: BRConfig,
614 category: str | None = None,
615 skip_ids: set[str] | None = None,
616 only_ids: list[str] | set[str] | None = None,
617 type_prefixes: set[str] | None = None,
618) -> list[IssueInfo]:
619 """Find all issues matching criteria.
621 Args:
622 config: Project configuration
623 category: Optional category to filter (e.g., "bugs")
624 skip_ids: Issue IDs to skip
625 only_ids: If provided, only include these issue IDs. When a list,
626 results are returned in list order (input sequence preserved).
627 When a set, results are sorted by priority as usual.
628 type_prefixes: If provided, only include issues whose ID starts with
629 one of these prefixes (e.g., {"BUG", "ENH"})
631 Returns:
632 List of IssueInfo sorted by priority, or in only_ids list order when
633 only_ids is a list
634 """
635 skip_ids = skip_ids or set()
636 parser = IssueParser(config)
637 issues: list[IssueInfo] = []
639 # Get completed and deferred directories for duplicate detection
640 completed_dir = config.get_completed_dir()
641 deferred_dir = config.get_deferred_dir()
643 # Pre-materialize filename sets once to avoid O(N) stat syscalls in the hot loop
644 completed_names = (
645 frozenset(p.name for p in completed_dir.glob("*.md"))
646 if completed_dir.exists()
647 else frozenset()
648 )
649 deferred_names = (
650 frozenset(p.name for p in deferred_dir.glob("*.md"))
651 if deferred_dir.exists()
652 else frozenset()
653 )
655 # Determine which categories to search
656 if category:
657 categories = [category] if category in config.issue_categories else []
658 else:
659 categories = config.issue_categories
661 for cat in categories:
662 issue_dir = config.get_issue_dir(cat)
663 if not issue_dir.exists():
664 continue
666 for issue_file in issue_dir.glob("*.md"):
667 # Pre-flight check: skip if already exists in completed or deferred directory
668 if issue_file.name in completed_names:
669 continue
670 if issue_file.name in deferred_names:
671 continue
673 info = parser.parse_file(issue_file)
674 # Apply skip filter
675 if info.issue_id in skip_ids:
676 continue
677 # Apply only filter (if specified)
678 if only_ids is not None and not any(_id_matches(info.issue_id, p) for p in only_ids):
679 continue
680 # Apply type filter (if specified)
681 if type_prefixes is not None:
682 prefix = info.issue_id.split("-", 1)[0]
683 if prefix not in type_prefixes:
684 continue
685 issues.append(info)
687 # When only_ids is a list, preserve input order; otherwise sort by priority
688 if isinstance(only_ids, list):
689 issues.sort(
690 key=lambda x: next(
691 (i for i, p in enumerate(only_ids) if _id_matches(x.issue_id, p)),
692 len(only_ids),
693 )
694 )
695 else:
696 issues.sort(key=lambda x: (x.priority_int, x.issue_id))
697 return issues
700def find_highest_priority_issue(
701 config: BRConfig,
702 category: str | None = None,
703 skip_ids: set[str] | None = None,
704 only_ids: set[str] | None = None,
705 type_prefixes: set[str] | None = None,
706) -> IssueInfo | None:
707 """Find the highest priority issue.
709 Args:
710 config: Project configuration
711 category: Optional category to filter
712 skip_ids: Issue IDs to skip
713 only_ids: If provided, only include these issue IDs
714 type_prefixes: If provided, only include issues with these type prefixes
716 Returns:
717 Highest priority IssueInfo or None if no issues found
718 """
719 issues = find_issues(config, category, skip_ids, only_ids, type_prefixes)
720 return issues[0] if issues else None