Coverage for little_loops / issue_parser.py: 24%

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 implementation_order_risk: bool | None = None 

271 learning_tests_required: list[str] | None = None 

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

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

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

275 milestone: str | None = None 

276 status: str = "open" 

277 

278 @property 

279 def priority_int(self) -> int: 

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

281 # Support P0-P5 priorities 

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

283 if match: 

284 return int(match.group(1)) 

285 return 99 # Unknown priority sorts last 

286 

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

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

289 return { 

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

291 "issue_type": self.issue_type, 

292 "priority": self.priority, 

293 "issue_id": self.issue_id, 

294 "title": self.title, 

295 "blocked_by": self.blocked_by, 

296 "blocks": self.blocks, 

297 "parent": self.parent, 

298 "depends_on": self.depends_on, 

299 "relates_to": self.relates_to, 

300 "duplicate_of": self.duplicate_of, 

301 "discovered_by": self.discovered_by, 

302 "epic": self.epic, 

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

304 "effort": self.effort, 

305 "impact": self.impact, 

306 "confidence_score": self.confidence_score, 

307 "outcome_confidence": self.outcome_confidence, 

308 "score_complexity": self.score_complexity, 

309 "score_test_coverage": self.score_test_coverage, 

310 "score_ambiguity": self.score_ambiguity, 

311 "score_change_surface": self.score_change_surface, 

312 "size": self.size, 

313 "testable": self.testable, 

314 "decision_needed": self.decision_needed, 

315 "missing_artifacts": self.missing_artifacts, 

316 "implementation_order_risk": self.implementation_order_risk, 

317 "learning_tests_required": self.learning_tests_required, 

318 "session_commands": self.session_commands, 

319 "session_command_counts": self.session_command_counts, 

320 "labels": self.labels, 

321 "milestone": self.milestone, 

322 "status": self.status, 

323 } 

324 

325 @classmethod 

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

327 """Create IssueInfo from dictionary.""" 

328 return cls( 

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

330 issue_type=data["issue_type"], 

331 priority=data["priority"], 

332 issue_id=data["issue_id"], 

333 title=data["title"], 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

355 implementation_order_risk=data.get("implementation_order_risk"), 

356 learning_tests_required=data.get("learning_tests_required"), 

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

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

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

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

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

362 ) 

363 

364 

365class IssueParser: 

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

367 

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

369 """ 

370 

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

372 """Initialize parser with project configuration. 

373 

374 Args: 

375 config: Project configuration 

376 """ 

377 self.config = config 

378 self._build_prefix_map() 

379 

380 def _build_prefix_map(self) -> None: 

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

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

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

384 self._prefix_to_category[category.prefix] = category_name 

385 

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

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

388 

389 Args: 

390 issue_path: Path to the issue markdown file 

391 

392 Returns: 

393 Parsed IssueInfo 

394 """ 

395 filename = issue_path.name 

396 

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

398 priority = self._parse_priority(filename) 

399 

400 # Parse issue type and ID from filename 

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

402 

403 # Read content once for all content-based parsing 

404 content = self._read_content(issue_path) 

405 

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

407 frontmatter = parse_frontmatter(content) 

408 discovered_by = frontmatter.get("discovered_by") 

409 epic = frontmatter.get("epic") 

410 size = frontmatter.get("size") 

411 product_impact = self._parse_product_impact(frontmatter) 

412 effort_raw = frontmatter.get("effort") 

413 impact_raw = frontmatter.get("impact") 

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

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

416 confidence_raw = frontmatter.get("confidence_score") 

417 outcome_raw = frontmatter.get("outcome_confidence") 

418 confidence_score = ( 

419 int(confidence_raw) 

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

421 else None 

422 ) 

423 outcome_confidence = ( 

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

425 ) 

426 complexity_raw = frontmatter.get("score_complexity") 

427 test_coverage_raw = frontmatter.get("score_test_coverage") 

428 ambiguity_raw = frontmatter.get("score_ambiguity") 

429 change_surface_raw = frontmatter.get("score_change_surface") 

430 score_complexity = ( 

431 int(complexity_raw) 

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

433 else None 

434 ) 

435 score_test_coverage = ( 

436 int(test_coverage_raw) 

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

438 else None 

439 ) 

440 score_ambiguity = ( 

441 int(ambiguity_raw) 

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

443 else None 

444 ) 

445 score_change_surface = ( 

446 int(change_surface_raw) 

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

448 else None 

449 ) 

450 testable_raw = frontmatter.get("testable") 

451 if isinstance(testable_raw, str): 

452 testable_value: bool | None = ( 

453 testable_raw.lower() == "true" 

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

455 else None 

456 ) 

457 else: 

458 testable_value = testable_raw 

459 

460 decision_needed_raw = frontmatter.get("decision_needed") 

461 if isinstance(decision_needed_raw, str): 

462 decision_needed_value: bool | None = ( 

463 decision_needed_raw.lower() == "true" 

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

465 else None 

466 ) 

467 else: 

468 decision_needed_value = decision_needed_raw 

469 

470 missing_artifacts_raw = frontmatter.get("missing_artifacts") 

471 if isinstance(missing_artifacts_raw, str): 

472 missing_artifacts_value: bool | None = ( 

473 missing_artifacts_raw.lower() == "true" 

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

475 else None 

476 ) 

477 else: 

478 missing_artifacts_value = missing_artifacts_raw 

479 

480 implementation_order_risk_raw = frontmatter.get("implementation_order_risk") 

481 if isinstance(implementation_order_risk_raw, str): 

482 implementation_order_risk_value: bool | None = ( 

483 implementation_order_risk_raw.lower() == "true" 

484 if implementation_order_risk_raw.lower() in ("true", "false") 

485 else None 

486 ) 

487 else: 

488 implementation_order_risk_value = implementation_order_risk_raw 

489 

490 learning_tests_raw = frontmatter.get("learning_tests_required") 

491 if isinstance(learning_tests_raw, str): 

492 learning_tests_required_value: list[str] | None = [ 

493 t.strip() for t in learning_tests_raw.split(",") if t.strip() 

494 ] or None 

495 elif isinstance(learning_tests_raw, list): 

496 learning_tests_required_value = [str(t) for t in learning_tests_raw] or None 

497 else: 

498 learning_tests_required_value = None 

499 

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

501 if status == "open" and frontmatter.get("completed_at"): 

502 status = "done" 

503 

504 parent = frontmatter.get("parent") 

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

506 logger.warning( 

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

508 issue_path.name, 

509 ) 

510 parent = alias_val 

511 

512 duplicate_of = frontmatter.get("duplicate_of") 

513 

514 relates_to: list[str] = [] 

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

516 logger.warning( 

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

518 issue_path.name, 

519 ) 

520 relates_to = ( 

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

522 if isinstance(alias_val, str) 

523 else list(alias_val) 

524 ) 

525 

526 depends_on: list[str] = [] 

527 

528 # Parse title: prefer frontmatter title: field, then markdown header, then filename stem 

529 title = frontmatter.get("title") or self._parse_title_from_content(content, issue_path) 

530 blocked_by = self._parse_blocked_by(content) 

531 blocks = self._parse_blocks(content) 

532 

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

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

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

536 for fm_key, body_ids in ( 

537 ("blocked_by", blocked_by), 

538 ("blocks", blocks), 

539 ("depends_on", depends_on), 

540 ("relates_to", relates_to), 

541 ): 

542 fm_val = frontmatter.get(fm_key) 

543 if not fm_val: 

544 continue 

545 fm_ids = ( 

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

547 if isinstance(fm_val, str) 

548 else list(fm_val) 

549 ) 

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

551 logger.warning( 

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

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

554 issue_path.name, 

555 fm_key, 

556 fm_ids, 

557 body_ids, 

558 ) 

559 body_ids.clear() 

560 body_ids.extend(fm_ids) 

561 elif not body_ids: 

562 body_ids.extend(fm_ids) 

563 

564 # Parse labels from frontmatter 

565 labels: list[str] = [] 

566 fm_labels = frontmatter.get("labels") 

567 if fm_labels: 

568 if isinstance(fm_labels, str): 

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

570 else: 

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

572 

573 # Parse milestone from frontmatter 

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

575 

576 # Parse session commands from ## Session Log section 

577 from little_loops.session_log import count_session_commands, parse_session_log 

578 

579 session_commands = parse_session_log(content) 

580 session_command_counts = count_session_commands(content) 

581 

582 return IssueInfo( 

583 path=issue_path, 

584 issue_type=issue_type, 

585 priority=priority, 

586 issue_id=issue_id, 

587 title=title, 

588 blocked_by=blocked_by, 

589 blocks=blocks, 

590 parent=parent, 

591 depends_on=depends_on, 

592 relates_to=relates_to, 

593 duplicate_of=duplicate_of, 

594 discovered_by=discovered_by, 

595 epic=epic, 

596 product_impact=product_impact, 

597 effort=effort, 

598 impact=impact, 

599 confidence_score=confidence_score, 

600 outcome_confidence=outcome_confidence, 

601 score_complexity=score_complexity, 

602 score_test_coverage=score_test_coverage, 

603 score_ambiguity=score_ambiguity, 

604 score_change_surface=score_change_surface, 

605 size=size, 

606 testable=testable_value, 

607 decision_needed=decision_needed_value, 

608 missing_artifacts=missing_artifacts_value, 

609 implementation_order_risk=implementation_order_risk_value, 

610 learning_tests_required=learning_tests_required_value, 

611 session_commands=session_commands, 

612 session_command_counts=session_command_counts, 

613 labels=labels, 

614 milestone=milestone, 

615 status=status, 

616 ) 

617 

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

619 """Extract priority from filename. 

620 

621 Args: 

622 filename: Issue filename 

623 

624 Returns: 

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

626 """ 

627 for priority in self.config.issue_priorities: 

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

629 return priority 

630 # Default to lowest priority if not found 

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

632 

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

634 """Get category name from issue prefix. 

635 

636 Args: 

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

638 

639 Returns: 

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

641 """ 

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

643 

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

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

646 

647 Args: 

648 filename: Issue filename 

649 issue_path: Full path to issue file 

650 

651 Returns: 

652 Tuple of (issue_type, issue_id) 

653 """ 

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

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

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

657 match = re.search(pattern, filename) 

658 if match: 

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

660 return category, issue_id 

661 

662 # Fall back to inferring category from directory. 

663 parent_name = issue_path.parent.name 

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

665 if parent_name == category_config.dir: 

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

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

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

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

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

671 if priority_match: 

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

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

674 return category_name, issue_id 

675 

676 # Last resort: use filename as ID 

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

678 

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

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

681 

682 Args: 

683 filename: Issue filename 

684 prefix: Issue prefix to use 

685 

686 Returns: 

687 Generated issue ID 

688 """ 

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

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

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

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

693 if numbers: 

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

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

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

697 category = self._get_category_for_prefix(prefix) 

698 next_num = get_next_issue_number(self.config, category) 

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

700 

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

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

703 

704 Args: 

705 issue_path: Path to issue file 

706 

707 Returns: 

708 File content or empty string on error 

709 """ 

710 try: 

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

712 except Exception as e: 

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

714 return "" 

715 

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

717 """Extract title from issue file content. 

718 

719 Args: 

720 content: Pre-read file content 

721 issue_path: Path to issue file (for fallback) 

722 

723 Returns: 

724 Issue title or filename stem as fallback 

725 """ 

726 if content: 

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

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

729 if match: 

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

731 # Try first header of any format 

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

733 if match: 

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

735 # Fall back to filename 

736 return issue_path.stem 

737 

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

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

740 

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

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

743 Skips content inside code fences. 

744 

745 Args: 

746 content: File content to parse 

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

748 

749 Returns: 

750 List of issue IDs found in the section 

751 """ 

752 if not content: 

753 return [] 

754 

755 # Strip code fences to avoid matching sections in examples 

756 content_without_code = self._strip_code_fences(content) 

757 

758 # Match section header case-insensitively 

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

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

761 if not match: 

762 return [] 

763 

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

765 start = match.end() 

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

767 if next_section: 

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

769 else: 

770 section_content = content_without_code[start:] 

771 

772 # Extract issue IDs from list items 

773 issue_ids = ISSUE_ID_PATTERN.findall(section_content) 

774 return issue_ids 

775 

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

777 """Remove code fence blocks from content. 

778 

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

780 line numbers while removing code fence content from parsing. 

781 

782 Args: 

783 content: File content 

784 

785 Returns: 

786 Content with code fence blocks replaced by empty lines 

787 """ 

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

789 result = [] 

790 in_fence = False 

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

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

793 in_fence = not in_fence 

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

795 elif in_fence: 

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

797 else: 

798 result.append(line) 

799 return "\n".join(result) 

800 

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

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

803 

804 Args: 

805 content: File content to parse 

806 

807 Returns: 

808 List of issue IDs that block this issue 

809 """ 

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

811 

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

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

814 

815 Args: 

816 content: File content to parse 

817 

818 Returns: 

819 List of issue IDs that this issue blocks 

820 """ 

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

822 

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

824 """Extract product impact from frontmatter. 

825 

826 Args: 

827 frontmatter: Dictionary of frontmatter fields 

828 

829 Returns: 

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

831 """ 

832 # Check if any product fields are present 

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

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

835 return None