Coverage for little_loops / issue_progress.py: 0%

1"""EPIC progress aggregation: child-issue status rollup and oldest-open detection.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from datetime import datetime 

7from typing import TYPE_CHECKING 

8 

9if TYPE_CHECKING: 

10 from little_loops.issue_parser import IssueInfo 

11 

12_ALL_STATUSES = frozenset({"open", "in_progress", "blocked", "done", "cancelled", "deferred"}) 

13_OPEN_STATUSES = frozenset({"open", "in_progress", "blocked"}) 

14_TERMINAL_STATUSES = frozenset({"done", "cancelled"}) 

15 

16 

17@dataclass 

18class EpicProgress: 

19 """Aggregate progress metrics for an EPIC computed from its children.""" 

20 

21 epic_id: str 

22 epic_title: str 

23 children: list[IssueInfo] 

24 by_status: dict[str, int] 

25 percent_done: float 

26 percent_blocked: float 

27 oldest_open: IssueInfo | None 

28 oldest_open_age_days: int | None 

29 

30 def to_dict(self) -> dict: 

31 """Serialize to a JSON-serializable dict for --format json output.""" 

32 result: dict = { 

33 "epic_id": self.epic_id, 

34 "epic_title": self.epic_title, 

35 "total": len(self.children), 

36 "by_status": self.by_status, 

37 "percent_done": round(self.percent_done, 1), 

38 "percent_blocked": round(self.percent_blocked, 1), 

39 } 

40 if self.oldest_open is not None: 

41 result["oldest_open"] = { 

42 "id": self.oldest_open.issue_id, 

43 "title": self.oldest_open.title, 

44 "age_days": self.oldest_open_age_days, 

45 } 

46 else: 

47 result["oldest_open"] = None 

48 return result 

49 

50 

51def _issue_age_days(issue: IssueInfo) -> int | None: 

52 """Return age in days for an issue using captured_at → discovered_date → mtime fallback.""" 

53 from little_loops.cli.issues.search import _parse_discovered_date 

54 

55 try: 

56 content = issue.path.read_text(encoding="utf-8") 

57 except OSError: 

58 return None 

59 

60 dt = _parse_discovered_date(content, issue.path) 

61 if dt is None: 

62 return None 

63 delta = datetime.now() - dt 

64 return max(0, delta.days) 

65 

66 

67def compute_epic_progress( 

68 epic_id: str, 

69 all_issues: list[IssueInfo], 

70) -> EpicProgress | None: 

71 """Compute progress aggregates for an EPIC from all loaded issues. 

72 

73 Resolution uses union of ``relates_to:`` (forward) + ``parent:`` (backward), deduplicated. 

74 All statuses (including done/cancelled/deferred) are included in totals. 

75 

76 Returns None when the EPIC ID is not found in all_issues. 

77 """ 

78 epic_id = epic_id.upper() 

79 

80 epic_matches = [i for i in all_issues if i.issue_id == epic_id] 

81 if not epic_matches: 

82 return None 

83 epic_info = epic_matches[0] 

84 

85 forward_ids: set[str] = set(epic_info.relates_to) 

86 backward_ids: set[str] = {i.issue_id for i in all_issues if i.parent == epic_id} 

87 child_ids = forward_ids | backward_ids 

88 

89 children = [i for i in all_issues if i.issue_id in child_ids] 

90 

91 by_status: dict[str, int] = {} 

92 for child in children: 

93 s = child.status 

94 by_status[s] = by_status.get(s, 0) + 1 

95 

96 total = len(children) 

97 done_count = sum(by_status.get(s, 0) for s in _TERMINAL_STATUSES) 

98 blocked_count = by_status.get("blocked", 0) 

99 

100 percent_done = (done_count / total * 100) if total > 0 else 0.0 

101 percent_blocked = (blocked_count / total * 100) if total > 0 else 0.0 

102 

103 open_children = [c for c in children if c.status in _OPEN_STATUSES] 

104 

105 oldest_open: IssueInfo | None = None 

106 oldest_open_age_days: int | None = None 

107 

108 if open_children: 

109 aged = [((_issue_age_days(c) or -1), c) for c in open_children] 

110 aged.sort(key=lambda x: -x[0]) 

111 best_age, oldest_open = aged[0] 

112 oldest_open_age_days = best_age if best_age >= 0 else None 

113 

114 return EpicProgress( 

115 epic_id=epic_id, 

116 epic_title=epic_info.title, 

117 children=children, 

118 by_status=by_status, 

119 percent_done=percent_done, 

120 percent_blocked=percent_blocked, 

121 oldest_open=oldest_open, 

122 oldest_open_age_days=oldest_open_age_days, 

123 )