Coverage for little_loops / issue_parser.py: 25%

1"""Issue file parsing for little-loops. 

2 

3Parses issue markdown files to extract metadata like priority, ID, type, and title. 

4""" 

5 

6from __future__ import annotations 

7 

8import logging 

9import re 

10from dataclasses import dataclass, field 

11from pathlib import Path 

12from typing import TYPE_CHECKING, Any 

13 

14from little_loops.cli_args import _id_matches 

15from little_loops.frontmatter import parse_frontmatter 

16 

17if TYPE_CHECKING: 

18 from little_loops.config import BRConfig 

19 

20 

21logger = logging.getLogger(__name__) 

22 

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) 

27 

28 

29_NORMALIZED_RE = re.compile(r"^P[0-5]-(BUG|FEAT|ENH|EPIC)-[0-9]{3,}-[a-z0-9-]+\.md$") 

30_ISSUE_TYPE_RE = re.compile(r"-(BUG|FEAT|ENH|EPIC)-") 

31 

32 

33def is_normalized(filename: str) -> bool: 

34 """Check whether an issue filename conforms to naming conventions. 

35 

36 Args: 

37 filename: The basename of the issue file (e.g. 'P2-BUG-010-my-issue.md'). 

38 

39 Returns: 

40 True if the filename matches ``^P[0-5]-(BUG|FEAT|ENH|EPIC)-[0-9]{3,}-[a-z0-9-]+\\.md$``. 

41 """ 

42 return bool(_NORMALIZED_RE.match(filename)) 

43 

44 

45def is_formatted(issue_path: Path, templates_dir: Path | None = None) -> bool: 

46 """Check whether an issue file has been formatted. 

47 

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). 

51 

52 Args: 

53 issue_path: Path to the issue markdown file. 

54 templates_dir: Optional override for the templates directory. 

55 

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 

63 

64 try: 

65 content = issue_path.read_text(encoding="utf-8") 

66 except Exception: 

67 return False 

68 

69 # Criterion 1: /ll:format-issue appears in the session log 

70 if "/ll:format-issue" in parse_session_log(content): 

71 return True 

72 

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) 

78 

79 try: 

80 sections_data = load_issue_sections(issue_type, templates_dir) 

81 except Exception: 

82 return False 

83 

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) 

91 

92 if not required: 

93 return True 

94 

95 headings = {m.strip() for m in re.findall(r"^##\s+(.+)$", content, re.MULTILINE)} 

96 return required.issubset(headings) 

97 

98 

99def slugify(text: str) -> str: 

100 """Convert text to slug format for filenames. 

101 

102 Args: 

103 text: Text to convert 

104 

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() 

111 

112 

113def get_next_issue_number(config: BRConfig, category: str | None = None) -> int: 

114 """Determine the next globally unique issue number. 

115 

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. 

119 

120 Args: 

121 config: Project configuration 

122 category: Unused, kept for backwards compatibility 

123 

124 Returns: 

125 Next available issue number (globally unique across all types) 

126 """ 

127 max_num = 0 

128 

129 # Get all known prefixes from configuration 

130 all_prefixes = [cat_config.prefix for cat_config in config.issues.categories.values()] 

131 

132 # Directories to scan: ALL category directories. Status (open/done/deferred) 

133 # now lives in frontmatter, so all issues — active and inactive — are in 

134 # their type dir. We still scan the legacy completed/ and deferred/ dirs 

135 # if they happen to exist (in-flight migration safety). 

136 dirs_to_scan: list[Path] = [] 

137 for cat_name in config.issues.categories: 

138 dirs_to_scan.append(config.get_issue_dir(cat_name)) 

139 legacy_completed = config.project_root / config.issues.base_dir / "completed" 

140 legacy_deferred = config.project_root / config.issues.base_dir / "deferred" 

141 if legacy_completed.exists(): 

142 dirs_to_scan.append(legacy_completed) 

143 if legacy_deferred.exists(): 

144 dirs_to_scan.append(legacy_deferred) 

145 

146 if not all_prefixes: 

147 return max_num + 1 

148 

149 # Pre-compile a single union regex to match any known prefix 

150 prefix_pattern = re.compile(r"(?:" + "|".join(re.escape(p) for p in all_prefixes) + r")-(\d+)") 

151 

152 for dir_path in dirs_to_scan: 

153 if not dir_path.exists(): 

154 continue 

155 for file in dir_path.glob("*.md"): 

156 match = prefix_pattern.search(file.name) 

157 if match: 

158 num = int(match.group(1)) 

159 if num > max_num: 

160 max_num = num 

161 

162 return max_num + 1 

163 

164 

165@dataclass 

166class ProductImpact: 

167 """Product impact assessment for an issue. 

168 

169 Attributes: 

170 goal_alignment: ID of the strategic priority this supports 

171 persona_impact: ID of the persona affected 

172 business_value: Business value assessment (high|medium|low) 

173 user_benefit: Description of how this helps the target user 

174 """ 

175 

176 goal_alignment: str | None = None 

177 persona_impact: str | None = None 

178 business_value: str | None = None # high|medium|low 

179 user_benefit: str | None = None 

180 

181 def to_dict(self) -> dict[str, Any]: 

182 """Convert to dictionary for JSON serialization.""" 

183 return { 

184 "goal_alignment": self.goal_alignment, 

185 "persona_impact": self.persona_impact, 

186 "business_value": self.business_value, 

187 "user_benefit": self.user_benefit, 

188 } 

189 

190 @classmethod 

191 def from_dict(cls, data: dict[str, Any] | None) -> ProductImpact | None: 

192 """Create ProductImpact from dictionary. 

193 

194 Args: 

195 data: Dictionary with product impact fields, or None 

196 

197 Returns: 

198 ProductImpact instance or None if data is None/empty 

199 """ 

200 if not data: 

201 return None 

202 return cls( 

203 goal_alignment=data.get("goal_alignment"), 

204 persona_impact=data.get("persona_impact"), 

205 business_value=data.get("business_value"), 

206 user_benefit=data.get("user_benefit"), 

207 ) 

208 

209 

210@dataclass 

211class IssueInfo: 

212 """Parsed information from an issue file. 

213 

214 Attributes: 

215 path: Path to the issue file 

216 issue_type: Type of issue (e.g., "bugs", "features") 

217 priority: Priority level (e.g., "P0", "P1") 

218 issue_id: Issue identifier (e.g., "BUG-123") 

219 title: Issue title from markdown header 

220 blocked_by: List of issue IDs that block this issue 

221 blocks: List of issue IDs that this issue blocks 

222 discovered_by: Source command/workflow that created this issue 

223 product_impact: Product impact assessment (optional) 

224 effort: Effort estimate (1=low, 2=medium, 3=high), inferred from priority if absent 

225 impact: Impact estimate (1=low, 2=medium, 3=high), inferred from priority if absent 

226 confidence_score: Readiness score (0-100) written by /ll:confidence-check, or None 

227 outcome_confidence: Outcome confidence (0-100) written by /ll:confidence-check, or None 

228 score_complexity: Outcome criterion A – Complexity (0-25), written by /ll:confidence-check, or None 

229 score_test_coverage: Outcome criterion B – Test Coverage (0-25), written by /ll:confidence-check, or None 

230 score_ambiguity: Outcome criterion C – Ambiguity (0-25), written by /ll:confidence-check, or None 

231 score_change_surface: Outcome criterion D – Change Surface (0-25), written by /ll:confidence-check, or None 

232 testable: Whether TDD phase should be applied; False skips TDD, None treated as testable 

233 session_commands: Distinct /ll:* commands found in the ## Session Log section 

234 session_command_counts: Per-command occurrence counts from the ## Session Log section 

235 labels: Labels extracted from the ## Labels section of the issue file 

236 milestone: Sprint or milestone name this issue is assigned to; None if unassigned 

237 status: Issue lifecycle status read from frontmatter; defaults to "open" 

238 parent: Parent issue ID (e.g., EPIC-123); populated from frontmatter `parent:` or deprecated `parent_issue:` 

239 depends_on: List of issue IDs this issue depends on (soft prerequisite) 

240 relates_to: List of related issue IDs; populated from frontmatter `relates_to:` or deprecated `related:` 

241 duplicate_of: Issue ID that this issue duplicates 

242 """ 

243 

244 path: Path 

245 issue_type: str 

246 priority: str 

247 issue_id: str 

248 title: str 

249 blocked_by: list[str] = field(default_factory=list) 

250 blocks: list[str] = field(default_factory=list) 

251 parent: str | None = None 

252 depends_on: list[str] = field(default_factory=list) 

253 relates_to: list[str] = field(default_factory=list) 

254 duplicate_of: str | None = None 

255 discovered_by: str | None = None 

256 epic: str | None = None 

257 product_impact: ProductImpact | None = None 

258 effort: int | None = None 

259 impact: int | None = None 

260 confidence_score: int | None = None 

261 outcome_confidence: int | None = None 

262 score_complexity: int | None = None 

263 score_test_coverage: int | None = None 

264 score_ambiguity: int | None = None 

265 score_change_surface: int | None = None 

266 size: str | None = None 

267 testable: bool | None = None 

268 decision_needed: bool | None = None 

269 missing_artifacts: bool | None = None 

270 session_commands: list[str] = field(default_factory=list) 

271 session_command_counts: dict[str, int] = field(default_factory=dict) 

272 labels: list[str] = field(default_factory=list) 

273 milestone: str | None = None 

274 status: str = "open" 

275 

276 @property 

277 def priority_int(self) -> int: 

278 """Convert priority to integer for comparison (lower = higher priority).""" 

279 # Support P0-P5 priorities 

280 match = re.match(r"^P(\d+)$", self.priority) 

281 if match: 

282 return int(match.group(1)) 

283 return 99 # Unknown priority sorts last 

284 

285 def to_dict(self) -> dict[str, Any]: 

286 """Convert to dictionary for JSON serialization.""" 

287 return { 

288 "path": str(self.path), 

289 "issue_type": self.issue_type, 

290 "priority": self.priority, 

291 "issue_id": self.issue_id, 

292 "title": self.title, 

293 "blocked_by": self.blocked_by, 

294 "blocks": self.blocks, 

295 "parent": self.parent, 

296 "depends_on": self.depends_on, 

297 "relates_to": self.relates_to, 

298 "duplicate_of": self.duplicate_of, 

299 "discovered_by": self.discovered_by, 

300 "epic": self.epic, 

301 "product_impact": (self.product_impact.to_dict() if self.product_impact else None), 

302 "effort": self.effort, 

303 "impact": self.impact, 

304 "confidence_score": self.confidence_score, 

305 "outcome_confidence": self.outcome_confidence, 

306 "score_complexity": self.score_complexity, 

307 "score_test_coverage": self.score_test_coverage, 

308 "score_ambiguity": self.score_ambiguity, 

309 "score_change_surface": self.score_change_surface, 

310 "size": self.size, 

311 "testable": self.testable, 

312 "decision_needed": self.decision_needed, 

313 "missing_artifacts": self.missing_artifacts, 

314 "session_commands": self.session_commands, 

315 "session_command_counts": self.session_command_counts, 

316 "labels": self.labels, 

317 "milestone": self.milestone, 

318 "status": self.status, 

319 } 

320 

321 @classmethod 

322 def from_dict(cls, data: dict[str, Any]) -> IssueInfo: 

323 """Create IssueInfo from dictionary.""" 

324 return cls( 

325 path=Path(data["path"]), 

326 issue_type=data["issue_type"], 

327 priority=data["priority"], 

328 issue_id=data["issue_id"], 

329 title=data["title"], 

330 blocked_by=data.get("blocked_by", []), 

331 blocks=data.get("blocks", []), 

332 parent=data.get("parent"), 

333 depends_on=data.get("depends_on", []), 

334 relates_to=data.get("relates_to", []), 

335 duplicate_of=data.get("duplicate_of"), 

336 discovered_by=data.get("discovered_by"), 

337 epic=data.get("epic"), 

338 product_impact=ProductImpact.from_dict(data.get("product_impact")), 

339 effort=data.get("effort"), 

340 impact=data.get("impact"), 

341 confidence_score=data.get("confidence_score"), 

342 outcome_confidence=data.get("outcome_confidence"), 

343 score_complexity=data.get("score_complexity"), 

344 score_test_coverage=data.get("score_test_coverage"), 

345 score_ambiguity=data.get("score_ambiguity"), 

346 score_change_surface=data.get("score_change_surface"), 

347 size=data.get("size"), 

348 testable=data.get("testable"), 

349 decision_needed=data.get("decision_needed"), 

350 missing_artifacts=data.get("missing_artifacts"), 

351 session_commands=data.get("session_commands", []), 

352 session_command_counts=data.get("session_command_counts", {}), 

353 labels=data.get("labels", []), 

354 milestone=data.get("milestone"), 

355 status=data.get("status", "open"), 

356 ) 

357 

358 

359class IssueParser: 

360 """Parses issue files based on project configuration. 

361 

362 Uses BRConfig to understand issue categories, prefixes, and priorities. 

363 """ 

364 

365 def __init__(self, config: BRConfig) -> None: 

366 """Initialize parser with project configuration. 

367 

368 Args: 

369 config: Project configuration 

370 """ 

371 self.config = config 

372 self._build_prefix_map() 

373 

374 def _build_prefix_map(self) -> None: 

375 """Build mapping from issue prefixes to category names.""" 

376 self._prefix_to_category: dict[str, str] = {} 

377 for category_name, category in self.config.issues.categories.items(): 

378 self._prefix_to_category[category.prefix] = category_name 

379 

380 def parse_file(self, issue_path: Path) -> IssueInfo: 

381 """Parse an issue file to extract metadata. 

382 

383 Args: 

384 issue_path: Path to the issue markdown file 

385 

386 Returns: 

387 Parsed IssueInfo 

388 """ 

389 filename = issue_path.name 

390 

391 # Parse priority from filename prefix (e.g., P1-BUG-123-...) 

392 priority = self._parse_priority(filename) 

393 

394 # Parse issue type and ID from filename 

395 issue_type, issue_id = self._parse_type_and_id(filename, issue_path) 

396 

397 # Read content once for all content-based parsing 

398 content = self._read_content(issue_path) 

399 

400 # Parse frontmatter for discovered_by, epic, product impact, effort, and impact 

401 frontmatter = parse_frontmatter(content) 

402 discovered_by = frontmatter.get("discovered_by") 

403 epic = frontmatter.get("epic") 

404 size = frontmatter.get("size") 

405 product_impact = self._parse_product_impact(frontmatter) 

406 effort_raw = frontmatter.get("effort") 

407 impact_raw = frontmatter.get("impact") 

408 effort = int(effort_raw) if effort_raw is not None and str(effort_raw).isdigit() else None 

409 impact = int(impact_raw) if impact_raw is not None and str(impact_raw).isdigit() else None 

410 confidence_raw = frontmatter.get("confidence_score") 

411 outcome_raw = frontmatter.get("outcome_confidence") 

412 confidence_score = ( 

413 int(confidence_raw) 

414 if confidence_raw is not None and str(confidence_raw).isdigit() 

415 else None 

416 ) 

417 outcome_confidence = ( 

418 int(outcome_raw) if outcome_raw is not None and str(outcome_raw).isdigit() else None 

419 ) 

420 complexity_raw = frontmatter.get("score_complexity") 

421 test_coverage_raw = frontmatter.get("score_test_coverage") 

422 ambiguity_raw = frontmatter.get("score_ambiguity") 

423 change_surface_raw = frontmatter.get("score_change_surface") 

424 score_complexity = ( 

425 int(complexity_raw) 

426 if complexity_raw is not None and str(complexity_raw).isdigit() 

427 else None 

428 ) 

429 score_test_coverage = ( 

430 int(test_coverage_raw) 

431 if test_coverage_raw is not None and str(test_coverage_raw).isdigit() 

432 else None 

433 ) 

434 score_ambiguity = ( 

435 int(ambiguity_raw) 

436 if ambiguity_raw is not None and str(ambiguity_raw).isdigit() 

437 else None 

438 ) 

439 score_change_surface = ( 

440 int(change_surface_raw) 

441 if change_surface_raw is not None and str(change_surface_raw).isdigit() 

442 else None 

443 ) 

444 testable_raw = frontmatter.get("testable") 

445 if isinstance(testable_raw, str): 

446 testable_value: bool | None = ( 

447 testable_raw.lower() == "true" 

448 if testable_raw.lower() in ("true", "false") 

449 else None 

450 ) 

451 else: 

452 testable_value = testable_raw 

453 

454 decision_needed_raw = frontmatter.get("decision_needed") 

455 if isinstance(decision_needed_raw, str): 

456 decision_needed_value: bool | None = ( 

457 decision_needed_raw.lower() == "true" 

458 if decision_needed_raw.lower() in ("true", "false") 

459 else None 

460 ) 

461 else: 

462 decision_needed_value = decision_needed_raw 

463 

464 missing_artifacts_raw = frontmatter.get("missing_artifacts") 

465 if isinstance(missing_artifacts_raw, str): 

466 missing_artifacts_value: bool | None = ( 

467 missing_artifacts_raw.lower() == "true" 

468 if missing_artifacts_raw.lower() in ("true", "false") 

469 else None 

470 ) 

471 else: 

472 missing_artifacts_value = missing_artifacts_raw 

473 

474 status = frontmatter.get("status", "open") 

475 

476 parent = frontmatter.get("parent") 

477 if parent is None and (alias_val := frontmatter.get("parent_issue")): 

478 logger.warning( 

479 "%s: deprecated frontmatter key 'parent_issue' — rename to 'parent'", 

480 issue_path.name, 

481 ) 

482 parent = alias_val 

483 

484 duplicate_of = frontmatter.get("duplicate_of") 

485 

486 relates_to: list[str] = [] 

487 if alias_val := frontmatter.get("related"): 

488 logger.warning( 

489 "%s: deprecated frontmatter key 'related' — rename to 'relates_to'", 

490 issue_path.name, 

491 ) 

492 relates_to = ( 

493 [id.strip() for id in alias_val.strip("\"'").split(",") if id.strip()] 

494 if isinstance(alias_val, str) 

495 else list(alias_val) 

496 ) 

497 

498 depends_on: list[str] = [] 

499 

500 # Parse title and dependencies from file content 

501 title = self._parse_title_from_content(content, issue_path) 

502 blocked_by = self._parse_blocked_by(content) 

503 blocks = self._parse_blocks(content) 

504 

505 # Also read blocked_by/blocks/depends_on/relates_to from frontmatter (canonical format). 

506 # When both sources provide values and they differ, prefer frontmatter and warn 

507 # so stale body sections are surfaced rather than silently merged. 

508 for fm_key, body_ids in ( 

509 ("blocked_by", blocked_by), 

510 ("blocks", blocks), 

511 ("depends_on", depends_on), 

512 ("relates_to", relates_to), 

513 ): 

514 fm_val = frontmatter.get(fm_key) 

515 if not fm_val: 

516 continue 

517 fm_ids = ( 

518 [id.strip() for id in fm_val.strip("\"'").split(",") if id.strip()] 

519 if isinstance(fm_val, str) 

520 else list(fm_val) 

521 ) 

522 if body_ids and set(fm_ids) != set(body_ids): 

523 logger.warning( 

524 "%s: frontmatter %s %s conflicts with body section %s; " 

525 "preferring frontmatter — update or remove the stale body section", 

526 issue_path.name, 

527 fm_key, 

528 fm_ids, 

529 body_ids, 

530 ) 

531 body_ids.clear() 

532 body_ids.extend(fm_ids) 

533 elif not body_ids: 

534 body_ids.extend(fm_ids) 

535 

536 # Parse labels from frontmatter 

537 labels: list[str] = [] 

538 fm_labels = frontmatter.get("labels") 

539 if fm_labels: 

540 if isinstance(fm_labels, str): 

541 labels = [lb.strip() for lb in fm_labels.split(",") if lb.strip()] 

542 else: 

543 labels = [str(lb) for lb in fm_labels] 

544 

545 # Parse milestone from frontmatter 

546 milestone: str | None = frontmatter.get("milestone") or None 

547 

548 # Parse session commands from ## Session Log section 

549 from little_loops.session_log import count_session_commands, parse_session_log 

550 

551 session_commands = parse_session_log(content) 

552 session_command_counts = count_session_commands(content) 

553 

554 return IssueInfo( 

555 path=issue_path, 

556 issue_type=issue_type, 

557 priority=priority, 

558 issue_id=issue_id, 

559 title=title, 

560 blocked_by=blocked_by, 

561 blocks=blocks, 

562 parent=parent, 

563 depends_on=depends_on, 

564 relates_to=relates_to, 

565 duplicate_of=duplicate_of, 

566 discovered_by=discovered_by, 

567 epic=epic, 

568 product_impact=product_impact, 

569 effort=effort, 

570 impact=impact, 

571 confidence_score=confidence_score, 

572 outcome_confidence=outcome_confidence, 

573 score_complexity=score_complexity, 

574 score_test_coverage=score_test_coverage, 

575 score_ambiguity=score_ambiguity, 

576 score_change_surface=score_change_surface, 

577 size=size, 

578 testable=testable_value, 

579 decision_needed=decision_needed_value, 

580 missing_artifacts=missing_artifacts_value, 

581 session_commands=session_commands, 

582 session_command_counts=session_command_counts, 

583 labels=labels, 

584 milestone=milestone, 

585 status=status, 

586 ) 

587 

588 def _parse_priority(self, filename: str) -> str: 

589 """Extract priority from filename. 

590 

591 Args: 

592 filename: Issue filename 

593 

594 Returns: 

595 Priority string (e.g., "P1") or last priority if not found 

596 """ 

597 for priority in self.config.issue_priorities: 

598 if filename.startswith(f"{priority}-"): 

599 return priority 

600 # Default to lowest priority if not found 

601 return self.config.issue_priorities[-1] if self.config.issue_priorities else "P3" 

602 

603 def _get_category_for_prefix(self, prefix: str) -> str: 

604 """Get category name from issue prefix. 

605 

606 Args: 

607 prefix: Issue prefix (e.g., "BUG", "FEAT") 

608 

609 Returns: 

610 Category name (e.g., "bugs", "features"), defaults to "bugs" 

611 """ 

612 return self._prefix_to_category.get(prefix, "bugs") 

613 

614 def _parse_type_and_id(self, filename: str, issue_path: Path) -> tuple[str, str]: 

615 """Extract issue type and ID from filename. 

616 

617 Args: 

618 filename: Issue filename 

619 issue_path: Full path to issue file 

620 

621 Returns: 

622 Tuple of (issue_type, issue_id) 

623 """ 

624 # Try to match known prefixes (BUG, FEAT, ENH, etc.) 

625 for prefix, category in self._prefix_to_category.items(): 

626 pattern = rf"({prefix})-(\d+)" 

627 match = re.search(pattern, filename) 

628 if match: 

629 issue_id = f"{match.group(1)}-{match.group(2)}" 

630 return category, issue_id 

631 

632 # Fall back to inferring category from directory. 

633 parent_name = issue_path.parent.name 

634 for category_name, category_config in self.config.issues.categories.items(): 

635 if parent_name == category_config.dir: 

636 # If the filename uses the standard P[0-5]-NNN-... shape but 

637 # omits the type token, capture the number directly and pair 

638 # it with the directory-derived prefix. Without this, generic 

639 # number scanning would pick up the priority digit instead. 

640 priority_match = re.match(r"^P\d+-(\d+)(?:[-.]|$)", filename) 

641 if priority_match: 

642 return category_name, f"{category_config.prefix}-{priority_match.group(1)}" 

643 issue_id = self._generate_id_from_filename(filename, category_config.prefix) 

644 return category_name, issue_id 

645 

646 # Last resort: use filename as ID 

647 return "bugs", filename.replace(".md", "") 

648 

649 def _generate_id_from_filename(self, filename: str, prefix: str) -> str: 

650 """Generate an issue ID from filename when not explicitly present. 

651 

652 Args: 

653 filename: Issue filename 

654 prefix: Issue prefix to use 

655 

656 Returns: 

657 Generated issue ID 

658 """ 

659 # Strip a leading priority token (e.g. "P2-") so it does not get 

660 # picked up as the issue number by the generic digit scan below. 

661 scan_target = re.sub(r"^P\d+-", "", filename) 

662 numbers = re.findall(r"\d+", scan_target) 

663 if numbers: 

664 return f"{prefix}-{numbers[0]}" 

665 # Use next sequential number instead of hash-based fallback 

666 # This ensures IDs are deterministic and don't collide with existing issues 

667 category = self._get_category_for_prefix(prefix) 

668 next_num = get_next_issue_number(self.config, category) 

669 return f"{prefix}-{next_num:03d}" 

670 

671 def _read_content(self, issue_path: Path) -> str: 

672 """Read file content, returning empty string on error. 

673 

674 Args: 

675 issue_path: Path to issue file 

676 

677 Returns: 

678 File content or empty string on error 

679 """ 

680 try: 

681 return issue_path.read_text(encoding="utf-8") 

682 except Exception as e: 

683 logger.warning("Failed to read %s: %s", issue_path.name, e) 

684 return "" 

685 

686 def _parse_title_from_content(self, content: str, issue_path: Path) -> str: 

687 """Extract title from issue file content. 

688 

689 Args: 

690 content: Pre-read file content 

691 issue_path: Path to issue file (for fallback) 

692 

693 Returns: 

694 Issue title or filename stem as fallback 

695 """ 

696 if content: 

697 # Look for markdown header: # ISSUE-ID: Title 

698 match = re.search(r"^#\s+[\w-]+:\s*(.+)$", content, re.MULTILINE) 

699 if match: 

700 return match.group(1).strip() 

701 # Try first header of any format 

702 match = re.search(r"^#\s+(.+)$", content, re.MULTILINE) 

703 if match: 

704 return match.group(1).strip() 

705 # Fall back to filename 

706 return issue_path.stem 

707 

708 def _parse_section_items(self, content: str, section_name: str) -> list[str]: 

709 """Extract issue IDs from a markdown section. 

710 

711 Finds section header (## Section Name) and extracts issue IDs 

712 from list items until the next section or end of file. 

713 Skips content inside code fences. 

714 

715 Args: 

716 content: File content to parse 

717 section_name: Section name to find (e.g., "Blocked By") 

718 

719 Returns: 

720 List of issue IDs found in the section 

721 """ 

722 if not content: 

723 return [] 

724 

725 # Strip code fences to avoid matching sections in examples 

726 content_without_code = self._strip_code_fences(content) 

727 

728 # Match section header case-insensitively 

729 section_pattern = rf"^##\s+{re.escape(section_name)}\s*$" 

730 match = re.search(section_pattern, content_without_code, re.MULTILINE | re.IGNORECASE) 

731 if not match: 

732 return [] 

733 

734 # Get content after section header until next ## header or end 

735 start = match.end() 

736 next_section = re.search(r"^##\s+", content_without_code[start:], re.MULTILINE) 

737 if next_section: 

738 section_content = content_without_code[start : start + next_section.start()] 

739 else: 

740 section_content = content_without_code[start:] 

741 

742 # Extract issue IDs from list items 

743 issue_ids = ISSUE_ID_PATTERN.findall(section_content) 

744 return issue_ids 

745 

746 def _strip_code_fences(self, content: str) -> str: 

747 """Remove code fence blocks from content. 

748 

749 Replaces content between ``` markers with empty lines to preserve 

750 line numbers while removing code fence content from parsing. 

751 

752 Args: 

753 content: File content 

754 

755 Returns: 

756 Content with code fence blocks replaced by empty lines 

757 """ 

758 # Match code fences: ``` or ```language through closing ``` 

759 result = [] 

760 in_fence = False 

761 for line in content.split("\n"): 

762 if line.startswith("```"): 

763 in_fence = not in_fence 

764 result.append("") # Preserve line count 

765 elif in_fence: 

766 result.append("") # Replace fenced content with empty line 

767 else: 

768 result.append(line) 

769 return "\n".join(result) 

770 

771 def _parse_blocked_by(self, content: str) -> list[str]: 

772 """Extract issue IDs from ## Blocked By section. 

773 

774 Args: 

775 content: File content to parse 

776 

777 Returns: 

778 List of issue IDs that block this issue 

779 """ 

780 return self._parse_section_items(content, "Blocked By") 

781 

782 def _parse_blocks(self, content: str) -> list[str]: 

783 """Extract issue IDs from ## Blocks section. 

784 

785 Args: 

786 content: File content to parse 

787 

788 Returns: 

789 List of issue IDs that this issue blocks 

790 """ 

791 return self._parse_section_items(content, "Blocks") 

792 

793 def _parse_product_impact(self, frontmatter: dict[str, Any]) -> ProductImpact | None: 

794 """Extract product impact from frontmatter. 

795 

796 Args: 

797 frontmatter: Dictionary of frontmatter fields 

798 

799 Returns: 

800 ProductImpact instance if any product fields are present, None otherwise 

801 """ 

802 # Check if any product fields are present 

803 product_fields = ("goal_alignment", "persona_impact", "business_value", "user_benefit") 

804 if not any(frontmatter.get(key) for key in product_fields): 

805 return None 

806 

807 return ProductImpact( 

808 goal_alignment=frontmatter.get("goal_alignment"), 

809 persona_impact=frontmatter.get("persona_impact"), 

810 business_value=frontmatter.get("business_value"), 

811 user_benefit=frontmatter.get("user_benefit"), 

812 ) 

813 

814 

815def find_issues( 

816 config: BRConfig, 

817 category: str | None = None, 

818 skip_ids: set[str] | None = None, 

819 only_ids: list[str] | set[str] | None = None, 

820 type_prefixes: set[str] | None = None, 

821) -> list[IssueInfo]: 

822 """Find all issues matching criteria. 

823 

824 Args: 

825 config: Project configuration 

826 category: Optional category to filter (e.g., "bugs") 

827 skip_ids: Issue IDs to skip 

828 only_ids: If provided, only include these issue IDs. When a list, 

829 results are returned in list order (input sequence preserved). 

830 When a set, results are sorted by priority as usual. 

831 type_prefixes: If provided, only include issues whose ID starts with 

832 one of these prefixes (e.g., {"BUG", "ENH"}) 

833 

834 Returns: 

835 List of IssueInfo sorted by priority, or in only_ids list order when 

836 only_ids is a list 

837 """ 

838 skip_ids = skip_ids or set() 

839 parser = IssueParser(config) 

840 issues: list[IssueInfo] = [] 

841 

842 # Determine which categories to search 

843 if category: 

844 categories = [category] if category in config.issue_categories else [] 

845 else: