Coverage for little_loops / sync.py: 15%

1"""GitHub Issues sync implementation for little-loops. 

2 

3Provides bidirectional sync between local .issues/ files and GitHub Issues. 

4""" 

5 

6from __future__ import annotations 

7 

8import difflib 

9import json 

10import re 

11import subprocess 

12from dataclasses import dataclass, field 

13from datetime import UTC, datetime 

14from pathlib import Path 

15from typing import TYPE_CHECKING, Any 

16 

17import yaml 

18 

19from little_loops.frontmatter import parse_frontmatter, strip_frontmatter 

20from little_loops.issue_parser import get_next_issue_number 

21from little_loops.issue_template import assemble_issue_markdown, load_issue_sections 

22 

23if TYPE_CHECKING: 

24 from little_loops.config import BRConfig 

25 from little_loops.logger import Logger 

26 

27 

28@dataclass 

29class SyncedIssue: 

30 """Represents an issue's sync state.""" 

31 

32 local_path: Path | None = None 

33 issue_id: str = "" 

34 github_number: int | None = None 

35 github_url: str = "" 

36 last_synced: str = "" 

37 local_changed: bool = False 

38 github_changed: bool = False 

39 

40 

41@dataclass 

42class SyncResult: 

43 """Result of a sync operation.""" 

44 

45 action: str # push, pull, status 

46 success: bool 

47 created: list[str] = field(default_factory=list) 

48 updated: list[str] = field(default_factory=list) 

49 skipped: list[str] = field(default_factory=list) 

50 failed: list[tuple[str, str]] = field(default_factory=list) # (issue_id, reason) 

51 errors: list[str] = field(default_factory=list) 

52 

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

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

55 return { 

56 "action": self.action, 

57 "success": self.success, 

58 "created": self.created, 

59 "updated": self.updated, 

60 "skipped": self.skipped, 

61 "failed": self.failed, 

62 "errors": self.errors, 

63 } 

64 

65 

66@dataclass 

67class SyncStatus: 

68 """Sync status overview.""" 

69 

70 provider: str 

71 repo: str 

72 local_total: int = 0 

73 local_synced: int = 0 

74 local_unsynced: int = 0 

75 github_total: int = 0 

76 github_only: int = 0 

77 github_error: str | None = None 

78 

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

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

81 return { 

82 "provider": self.provider, 

83 "repo": self.repo, 

84 "local_total": self.local_total, 

85 "local_synced": self.local_synced, 

86 "local_unsynced": self.local_unsynced, 

87 "github_total": self.github_total, 

88 "github_only": self.github_only, 

89 "github_error": self.github_error, 

90 } 

91 

92 

93# ============================================================================= 

94# Helper Functions 

95# ============================================================================= 

96 

97 

98def _run_gh_command( 

99 args: list[str], 

100 logger: Logger, 

101 check: bool = True, 

102) -> subprocess.CompletedProcess[str]: 

103 """Run a gh CLI command and return result. 

104 

105 Args: 

106 args: Arguments to pass to gh CLI (e.g., ["issue", "list", "--json", "number"]) 

107 logger: Logger for output 

108 check: Whether to raise on non-zero exit (default True) 

109 

110 Returns: 

111 CompletedProcess with stdout/stderr 

112 

113 Raises: 

114 subprocess.CalledProcessError: If command fails and check=True 

115 """ 

116 cmd = ["gh"] + args 

117 logger.debug(f"Running: {' '.join(cmd)}") 

118 result = subprocess.run( 

119 cmd, 

120 capture_output=True, 

121 text=True, 

122 check=check, 

123 ) 

124 return result 

125 

126 

127def _check_gh_auth(logger: Logger) -> bool: 

128 """Check if gh CLI is authenticated. 

129 

130 Returns: 

131 True if authenticated, False otherwise 

132 """ 

133 try: 

134 result = _run_gh_command(["auth", "status"], logger, check=False) 

135 return result.returncode == 0 

136 except FileNotFoundError: 

137 logger.error("gh CLI not found. Install with: brew install gh") 

138 return False 

139 

140 

141def _get_repo_name(logger: Logger) -> str | None: 

142 """Get current repository name from gh CLI. 

143 

144 Returns: 

145 Repository name in owner/repo format, or None if not in a repo 

146 """ 

147 try: 

148 result = _run_gh_command( 

149 ["repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"], 

150 logger, 

151 check=False, 

152 ) 

153 if result.returncode == 0: 

154 return result.stdout.strip() 

155 except Exception as e: 

156 logger.debug(f"Could not get repo name: {e}") 

157 return None 

158 

159 

160def _update_issue_frontmatter( 

161 content: str, 

162 updates: dict[str, str | int], 

163) -> str: 

164 """Update or add frontmatter fields in issue content. 

165 

166 Args: 

167 content: Full file content 

168 updates: Fields to add/update in frontmatter 

169 

170 Returns: 

171 Updated content with modified frontmatter 

172 """ 

173 fm_match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL) 

174 if not fm_match: 

175 # No existing frontmatter, create it 

176 fm_text = yaml.dump(dict(updates), default_flow_style=False, sort_keys=False).strip() 

177 return f"---\n{fm_text}\n---\n{content}" 

178 

179 existing: dict[str, Any] = yaml.safe_load(fm_match.group(1)) or {} 

180 existing.update(updates) 

181 fm_text = yaml.dump(existing, default_flow_style=False, sort_keys=False).strip() 

182 return f"---\n{fm_text}\n---{content[fm_match.end() :]}" 

183 

184 

185def _parse_issue_title(content: str) -> str: 

186 """Extract title from issue content (after frontmatter). 

187 

188 Looks for first markdown heading: # ISSUE-ID: Title 

189 

190 Args: 

191 content: Full file content 

192 

193 Returns: 

194 Title string or empty string if not found 

195 """ 

196 content = strip_frontmatter(content) 

197 

198 # Find first heading 

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

200 line = line.strip() 

201 if line.startswith("# "): 

202 # Remove issue ID prefix if present 

203 title = line[2:].strip() 

204 # Pattern: ISSUE-ID: Title 

205 if ":" in title: 

206 parts = title.split(":", 1) 

207 if re.match(r"^[A-Z]+-\d+$", parts[0].strip()): 

208 return parts[1].strip() 

209 return title 

210 return "" 

211 

212 

213def _get_issue_body(content: str) -> str: 

214 """Extract body from issue content (after frontmatter and title). 

215 

216 Args: 

217 content: Full file content 

218 

219 Returns: 

220 Body content 

221 """ 

222 content = strip_frontmatter(content) 

223 

224 # Skip leading blank lines 

225 lines = content.split("\n") 

226 while lines and not lines[0].strip(): 

227 lines.pop(0) 

228 

229 # Skip title line 

230 if lines and lines[0].startswith("# "): 

231 lines.pop(0) 

232 

233 return "\n".join(lines).strip() 

234 

235 

236# ============================================================================= 

237# GitHubSyncManager Class 

238# ============================================================================= 

239 

240 

241class GitHubSyncManager: 

242 """Manages bidirectional sync between local issues and GitHub Issues.""" 

243 

244 def __init__( 

245 self, 

246 config: BRConfig, 

247 logger: Logger, 

248 dry_run: bool = False, 

249 ) -> None: 

250 """Initialize sync manager. 

251 

252 Args: 

253 config: Project configuration 

254 logger: Logger for output 

255 dry_run: If True, show what would be done without making changes 

256 """ 

257 self.config = config 

258 self.sync_config = config.sync 

259 self.logger = logger 

260 self.dry_run = dry_run 

261 self.issues_dir = config.project_root / config.issues.base_dir 

262 self._sections_data: dict[str, dict[str, Any]] = {} 

263 

264 def _get_local_issues(self) -> list[Path]: 

265 """Get all local issue files to sync. 

266 

267 Returns: 

268 List of issue file paths 

269 """ 

270 issues: list[Path] = [] 

271 for category in self.config.issue_categories: 

272 category_dir = self.config.get_issue_dir(category) 

273 if category_dir.exists(): 

274 for issue_file in category_dir.glob("*.md"): 

275 issues.append(issue_file) 

276 

277 # Include completed if configured 

278 if self.sync_config.github.sync_completed: 

279 completed_dir = self.config.get_completed_dir() 

280 if completed_dir.exists(): 

281 for issue_file in completed_dir.glob("*.md"): 

282 issues.append(issue_file) 

283 

284 return issues 

285 

286 def _extract_issue_id(self, filename: str) -> str: 

287 """Extract issue ID from filename. 

288 

289 Args: 

290 filename: Issue filename (e.g., P1-BUG-123-description.md) 

291 

292 Returns: 

293 Issue ID (e.g., BUG-123) 

294 """ 

295 # Pattern: P[0-5]-TYPE-NNN-description.md 

296 match = re.search(r"(BUG|FEAT|ENH)-(\d+)", filename) 

297 if match: 

298 return f"{match.group(1)}-{match.group(2)}" 

299 return "" 

300 

301 def _get_labels_for_issue(self, issue_path: Path) -> list[str]: 

302 """Determine GitHub labels for an issue. 

303 

304 Args: 

305 issue_path: Path to issue file 

306 

307 Returns: 

308 List of label names 

309 """ 

310 labels: list[str] = [] 

311 filename = issue_path.name 

312 

313 # Get type label from mapping 

314 issue_id = self._extract_issue_id(filename) 

315 if issue_id: 

316 type_prefix = issue_id.split("-")[0] 

317 type_label = self.sync_config.github.label_mapping.get(type_prefix) 

318 if type_label: 

319 labels.append(type_label) 

320 

321 # Add priority label if configured 

322 if self.sync_config.github.priority_labels: 

323 priority_match = re.match(r"^(P[0-5])-", filename) 

324 if priority_match: 

325 labels.append(priority_match.group(1).lower()) 

326 

327 return labels 

328 

329 def push_issues(self, issue_ids: list[str] | None = None) -> SyncResult: 

330 """Push local issues to GitHub. 

331 

332 Args: 

333 issue_ids: Specific issue IDs to push, or None for all 

334 

335 Returns: 

336 SyncResult with operation details 

337 """ 

338 result = SyncResult(action="push", success=True) 

339 

340 # Verify gh auth 

341 if not _check_gh_auth(self.logger): 

342 result.success = False 

343 result.errors.append("GitHub CLI not authenticated. Run: gh auth login") 

344 return result 

345 

346 # Get repo name 

347 repo = self.sync_config.github.repo or _get_repo_name(self.logger) 

348 if not repo: 

349 result.success = False 

350 result.errors.append("Could not determine repository. Set sync.github.repo in config.") 

351 return result 

352 

353 local_issues = self._get_local_issues() 

354 self.logger.info(f"Found {len(local_issues)} local issues") 

355 

356 for issue_path in local_issues: 

357 issue_id = self._extract_issue_id(issue_path.name) 

358 if not issue_id: 

359 self.logger.debug(f"Skipping {issue_path.name}: no issue ID found") 

360 continue 

361 

362 # Filter by issue_ids if specified 

363 if issue_ids and issue_id not in issue_ids: 

364 continue 

365 

366 try: 

367 self._push_single_issue(issue_path, issue_id, result) 

368 except Exception as e: 

369 result.failed.append((issue_id, str(e))) 

370 self.logger.error(f"Failed to push {issue_id}: {e}") 

371 

372 if result.failed: 

373 result.success = False 

374 

375 return result 

376 

377 def _push_single_issue( 

378 self, 

379 issue_path: Path, 

380 issue_id: str, 

381 result: SyncResult, 

382 ) -> None: 

383 """Push a single issue to GitHub. 

384 

385 Args: 

386 issue_path: Path to local issue file 

387 issue_id: Issue ID (e.g., BUG-123) 

388 result: SyncResult to update 

389 """ 

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

391 frontmatter = parse_frontmatter(content, coerce_types=True) 

392 title = _parse_issue_title(content) 

393 body = _get_issue_body(content) 

394 

395 # Build full title with issue ID 

396 full_title = f"{issue_id}: {title}" if title else issue_id 

397 

398 # Get labels 

399 labels = self._get_labels_for_issue(issue_path) 

400 

401 github_number = frontmatter.get("github_issue") 

402 

403 if self.dry_run: 

404 if github_number: 

405 result.updated.append(f"{issue_id} → #{github_number} (would update)") 

406 self.logger.info(f"Would update GitHub issue #{github_number} for {issue_id}") 

407 else: 

408 result.created.append(f"{issue_id} (would create)") 

409 self.logger.info(f"Would create GitHub issue for {issue_id}") 

410 return 

411 

412 if github_number: 

413 # Update existing issue 

414 self._update_github_issue(int(github_number), full_title, body, issue_id, result) 

415 else: 

416 # Create new issue 

417 new_number = self._create_github_issue(full_title, body, labels, issue_id, result) 

418 if new_number: 

419 # Update local frontmatter 

420 self._update_local_frontmatter(issue_path, content, new_number) 

421 

422 def _create_github_issue( 

423 self, 

424 title: str, 

425 body: str, 

426 labels: list[str], 

427 issue_id: str, 

428 result: SyncResult, 

429 ) -> int | None: 

430 """Create a new GitHub issue. 

431 

432 Returns: 

433 GitHub issue number if successful, None otherwise 

434 """ 

435 args = ["issue", "create", "--title", title, "--body", body] 

436 for label in labels: 

437 args.extend(["--label", label]) 

438 

439 try: 

440 cmd_result = _run_gh_command(args, self.logger) 

441 # gh issue create outputs the URL 

442 url = cmd_result.stdout.strip() 

443 # Extract issue number from URL 

444 match = re.search(r"/issues/(\d+)$", url) 

445 if match: 

446 issue_num = int(match.group(1)) 

447 result.created.append(f"{issue_id} → #{issue_num}") 

448 self.logger.success(f"Created GitHub issue #{issue_num} for {issue_id}") 

449 return issue_num 

450 except subprocess.CalledProcessError as e: 

451 result.failed.append((issue_id, f"gh issue create failed: {e.stderr}")) 

452 self.logger.error(f"Failed to create GitHub issue for {issue_id}: {e.stderr}") 

453 return None 

454 

455 def _update_github_issue( 

456 self, 

457 github_number: int, 

458 title: str, 

459 body: str, 

460 issue_id: str, 

461 result: SyncResult, 

462 ) -> None: 

463 """Update an existing GitHub issue.""" 

464 args = [ 

465 "issue", 

466 "edit", 

467 str(github_number), 

468 "--title", 

469 title, 

470 "--body", 

471 body, 

472 ] 

473 try: 

474 _run_gh_command(args, self.logger) 

475 result.updated.append(f"{issue_id} → #{github_number}") 

476 self.logger.success(f"Updated GitHub issue #{github_number} for {issue_id}") 

477 except subprocess.CalledProcessError as e: 

478 result.failed.append((issue_id, f"gh issue edit failed: {e.stderr}")) 

479 self.logger.error(f"Failed to update GitHub issue #{github_number}: {e.stderr}") 

480 

481 def _update_local_frontmatter( 

482 self, 

483 issue_path: Path, 

484 content: str, 

485 github_number: int, 

486 ) -> None: 

487 """Update local issue file with GitHub sync info.""" 

488 repo = self.sync_config.github.repo or _get_repo_name(self.logger) or "" 

489 github_url = f"https://github.com/{repo}/issues/{github_number}" if repo else "" 

490 now = datetime.now(UTC).isoformat(timespec="seconds") 

491 

492 updates: dict[str, str | int] = { 

493 "github_issue": github_number, 

494 "github_url": github_url, 

495 "last_synced": now, 

496 } 

497 updated_content = _update_issue_frontmatter(content, updates) 

498 issue_path.write_text(updated_content, encoding="utf-8") 

499 self.logger.debug(f"Updated frontmatter in {issue_path.name}") 

500 

501 def pull_issues(self, labels: list[str] | None = None) -> SyncResult: 

502 """Pull GitHub Issues to local files. 

503 

504 Args: 

505 labels: Filter by labels, or None for all recognized labels 

506 

507 Returns: 

508 SyncResult with operation details 

509 """ 

510 result = SyncResult(action="pull", success=True) 

511 

512 # Verify gh auth 

513 if not _check_gh_auth(self.logger): 

514 result.success = False 

515 result.errors.append("GitHub CLI not authenticated. Run: gh auth login") 

516 return result 

517 

518 # List GitHub issues 

519 try: 

520 gh_args = [ 

521 "issue", 

522 "list", 

523 "--json", 

524 "number,title,body,labels,state,url", 

525 "--limit", 

526 "100", 

527 ] 

528 if labels: 

529 for label in labels: 

530 gh_args.extend(["--label", label]) 

531 cmd_result = _run_gh_command(gh_args, self.logger) 

532 github_issues = json.loads(cmd_result.stdout) 

533 except Exception as e: 

534 result.success = False 

535 result.errors.append(f"Failed to list GitHub issues: {e}") 

536 return result 

537 

538 # Get existing local issue IDs 

539 local_github_numbers = self._get_local_github_numbers() 

540 

541 for gh_issue in github_issues: 

542 gh_number = gh_issue["number"] 

543 gh_state = gh_issue.get("state", "OPEN") 

544 

545 # Skip closed issues unless configured 

546 if gh_state != "OPEN" and not self.sync_config.github.sync_completed: 

547 result.skipped.append(f"#{gh_number} (closed)") 

548 continue 

549 

550 # Skip if already tracked locally 

551 if gh_number in local_github_numbers: 

552 result.skipped.append(f"#{gh_number} (already tracked)") 

553 continue 

554 

555 # Check if has recognized labels 

556 gh_labels = [lbl.get("name", "") for lbl in gh_issue.get("labels", [])] 

557 issue_type = self._determine_issue_type(gh_labels) 

558 if not issue_type: 

559 result.skipped.append(f"#{gh_number} (no recognized type label)") 

560 continue 

561 

562 if self.dry_run: 

563 gh_title = gh_issue.get("title", f"Issue #{gh_number}") 

564 result.created.append(f"#{gh_number}: {gh_title} (would create as {issue_type})") 

565 self.logger.info(f"Would create local issue from GitHub #{gh_number}: {gh_title}") 

566 else: 

567 try: 

568 self._create_local_issue(gh_issue, issue_type, result) 

569 except Exception as e: 

570 result.failed.append((f"#{gh_number}", str(e))) 

571 

572 if result.failed: 

573 result.success = False 

574 

575 return result 

576 

577 def _get_local_github_numbers(self) -> set[int]: 

578 """Get set of GitHub issue numbers tracked locally.""" 

579 numbers: set[int] = set() 

580 for issue_path in self._get_local_issues(): 

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

582 frontmatter = parse_frontmatter(content, coerce_types=True) 

583 gh_num = frontmatter.get("github_issue") 

584 if gh_num is not None: 

585 try: 

586 numbers.add(int(gh_num)) 

587 except (ValueError, TypeError): 

588 self.logger.warning( 

589 f"Malformed github_issue value in {issue_path.name}: {gh_num!r}" 

590 ) 

591 return numbers 

592 

593 def _determine_issue_type(self, labels: list[str]) -> str | None: 

594 """Determine issue type from GitHub labels. 

595 

596 Returns: 

597 Issue type prefix (BUG, FEAT, ENH) or None 

598 """ 

599 # Reverse lookup from label_mapping 

600 reverse_map: dict[str, str] = {} 

601 for type_prefix, label in self.sync_config.github.label_mapping.items(): 

602 if label not in reverse_map: 

603 reverse_map[label] = type_prefix 

604 

605 for label in labels: 

606 if label in reverse_map: 

607 return reverse_map[label] 

608 return None 

609 

610 def _create_local_issue( 

611 self, 

612 gh_issue: dict[str, Any], 

613 issue_type: str, 

614 result: SyncResult, 

615 ) -> None: 

616 """Create a local issue file from GitHub issue.""" 

617 gh_number = gh_issue["number"] 

618 gh_title = gh_issue.get("title", f"Issue #{gh_number}") 

619 gh_body = gh_issue.get("body", "") or "" 

620 gh_url = gh_issue.get("url", "") 

621 gh_labels = [lbl.get("name", "") for lbl in gh_issue.get("labels", [])] 

622 

623 # Determine priority from labels or default 

624 priority = "P3" 

625 for label in gh_labels: 

626 if re.match(r"^p[0-5]$", label, re.IGNORECASE): 

627 priority = label.upper() 

628 break 

629 

630 # Generate next issue number (uses global numbering across all dirs) 

631 next_num = get_next_issue_number(self.config) 

632 

633 # Generate slug from title 

634 slug = re.sub(r"[^a-z0-9]+", "-", gh_title.lower())[:40].strip("-") 

635 

636 issue_id = f"{issue_type}-{next_num}" 

637 filename = f"{priority}-{issue_id}-{slug}.md" 

638 

639 # Determine category directory 

640 cat = self.config.issues.get_category_by_prefix(issue_type) 

641 category = cat.dir if cat else "features" 

642 category_dir = self.config.get_issue_dir(category) 

643 category_dir.mkdir(parents=True, exist_ok=True) 

644 

645 issue_path = category_dir / filename 

646 

647 # Build content using per-type sections template 

648 now = datetime.now(UTC).isoformat(timespec="seconds") 

649 today = datetime.now(UTC).strftime("%Y-%m-%d") 

650 

651 if issue_type not in self._sections_data: 

652 templates_dir = ( 

653 Path(self.config.issues.templates_dir) if self.config.issues.templates_dir else None 

654 ) 

655 self._sections_data[issue_type] = load_issue_sections(issue_type, templates_dir) 

656 

657 frontmatter = { 

658 "github_issue": gh_number, 

659 "github_url": gh_url, 

660 "last_synced": now, 

661 "discovered_by": "github_sync", 

662 "discovered_date": today, 

663 } 

664 section_content: dict[str, str] = {} 

665 if gh_body: 

666 section_content["Summary"] = gh_body 

667 section_content["Impact"] = ( 

668 f"- **Priority**: {priority}\n" 

669 f"- **Effort**: Unknown\n" 

670 f"- **Risk**: Unknown\n" 

671 f"- **Breaking Change**: Unknown" 

672 ) 

673 section_content["Status"] = f"**Open** | Created: {today} | Priority: {priority}" 

674 

675 labels_str = ", ".join(f"`{lbl}`" for lbl in gh_labels) if gh_labels else "" 

676 if labels_str: 

677 section_content["Labels"] = labels_str 

678 

679 variant = self.sync_config.github.pull_template 

680 content = assemble_issue_markdown( 

681 sections_data=self._sections_data[issue_type], 

682 issue_type=issue_type, 

683 variant=variant, 

684 issue_id=issue_id, 

685 title=gh_title, 

686 frontmatter=frontmatter, 

687 content=section_content, 

688 labels=gh_labels, 

689 ) 

690 issue_path.write_text(content, encoding="utf-8") 

691 result.created.append(f"#{gh_number} → {issue_id}") 

692 self.logger.success(f"Created {filename} from GitHub #{gh_number}") 

693 

694 def get_status(self) -> SyncStatus: 

695 """Get sync status overview. 

696 

697 Returns: 

698 SyncStatus with counts 

699 """ 

700 repo = self.sync_config.github.repo or _get_repo_name(self.logger) or "unknown" 

701 

702 status = SyncStatus( 

703 provider=self.sync_config.provider, 

704 repo=repo, 

705 ) 

706 

707 # Count local issues 

708 local_issues = self._get_local_issues() 

709 status.local_total = len(local_issues) 

710 

711 # Count synced (have github_issue) 

712 local_github_numbers = self._get_local_github_numbers() 

713 status.local_synced = len(local_github_numbers) 

714 status.local_unsynced = status.local_total - status.local_synced 

715 

716 # Count GitHub issues 

717 if _check_gh_auth(self.logger): 

718 try: 

719 cmd_result = _run_gh_command( 

720 ["issue", "list", "--json", "number", "--limit", "500"], 

721 self.logger, 

722 ) 

723 github_issues = json.loads(cmd_result.stdout) 

724 status.github_total = len(github_issues) 

725 

726 github_numbers = {issue["number"] for issue in github_issues} 

727 status.github_only = len(github_numbers - local_github_numbers) 

728 except Exception as e: 

729 status.github_error = f"Failed to query GitHub: {e}" 

730 self.logger.warning(status.github_error) 

731 

732 return status 

733 

734 def _find_local_issue(self, issue_id: str) -> Path | None: 

735 """Find the local file matching an issue ID. 

736 

737 Searches active category directories and completed directory. 

738 

739 Args: 

740 issue_id: Issue ID to find (e.g., BUG-123) 

741 

742 Returns: 

743 Path to the issue file, or None if not found 

744 """ 

745 for issue_path in self._get_local_issues(): 

746 if self._extract_issue_id(issue_path.name) == issue_id: 

747 return issue_path 

748 # Also check completed directory 

749 completed_dir = self.config.get_completed_dir() 

750 if completed_dir.exists(): 

751 for issue_file in completed_dir.glob("*.md"): 

752 if self._extract_issue_id(issue_file.name) == issue_id: 

753 return issue_file 

754 return None 

755 

756 def diff_issue(self, issue_id: str) -> SyncResult: 

757 """Show content differences between a local issue and its GitHub counterpart. 

758 

759 Args: 

760 issue_id: Issue ID to diff (e.g., BUG-123) 

761 

762 Returns: 

763 SyncResult with diff information 

764 """ 

765 result = SyncResult(action="diff", success=True) 

766 

767 if not _check_gh_auth(self.logger): 

768 result.success = False 

769 result.errors.append("GitHub CLI not authenticated. Run: gh auth login") 

770 return result 

771 

772 issue_path = self._find_local_issue(issue_id) 

773 if not issue_path: 

774 result.success = False 

775 result.errors.append(f"Local issue {issue_id} not found") 

776 return result 

777 

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

779 frontmatter = parse_frontmatter(content, coerce_types=True) 

780 github_number = frontmatter.get("github_issue") 

781 

782 if github_number is None: 

783 result.success = False 

784 result.errors.append( 

785 f"{issue_id} is not synced to GitHub (no github_issue in frontmatter)" 

786 ) 

787 return result 

788 

789 try: 

790 cmd_result = _run_gh_command( 

791 ["issue", "view", str(int(github_number)), "--json", "body", "-q", ".body"], 

792 self.logger, 

793 ) 

794 github_body = cmd_result.stdout.rstrip("\n") 

795 except subprocess.CalledProcessError as e: 

796 result.success = False 

797 result.errors.append(f"Failed to fetch GitHub issue #{github_number}: {e.stderr}") 

798 return result 

799 

800 local_body = _get_issue_body(content) 

801 

802 local_lines = local_body.splitlines(keepends=True) 

803 github_lines = github_body.splitlines(keepends=True) 

804 

805 diff = list( 

806 difflib.unified_diff( 

807 github_lines, 

808 local_lines, 

809 fromfile=f"github:#{github_number}", 

810 tofile=f"local:{issue_id}", 

811 ) 

812 ) 

813 

814 if diff: 

815 result.updated.append(f"{issue_id} (#{github_number}): differs") 

816 # Store diff lines in created field for display 

817 result.created = [line.rstrip("\n") for line in diff] 

818 else: 

819 result.skipped.append(f"{issue_id} (#{github_number}): in sync")