Coverage for little_loops / issue_progress.py: 0%
60 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""EPIC progress aggregation: child-issue status rollup and oldest-open detection."""
3from __future__ import annotations
5from dataclasses import dataclass
6from datetime import datetime
7from typing import TYPE_CHECKING
9if TYPE_CHECKING:
10 from little_loops.issue_parser import IssueInfo
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"})
17@dataclass
18class EpicProgress:
19 """Aggregate progress metrics for an EPIC computed from its children."""
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
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
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
55 try:
56 content = issue.path.read_text(encoding="utf-8")
57 except OSError:
58 return None
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)
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.
73 Child resolution uses only the ``parent:`` back-reference on child issues.
74 ``relates_to:`` is a cross-reference field (siblings, dependencies) and is
75 intentionally excluded to avoid inflating counts with non-child references.
76 All statuses (including done/cancelled/deferred) are included in totals.
78 Returns None when the EPIC ID is not found in all_issues.
79 """
80 epic_id = epic_id.upper()
82 epic_matches = [i for i in all_issues if i.issue_id == epic_id]
83 if not epic_matches:
84 return None
85 epic_info = epic_matches[0]
87 child_ids: set[str] = {i.issue_id for i in all_issues if i.parent == epic_id}
89 children = [i for i in all_issues if i.issue_id in child_ids]
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
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)
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
103 open_children = [c for c in children if c.status in _OPEN_STATUSES]
105 oldest_open: IssueInfo | None = None
106 oldest_open_age_days: int | None = None
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
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 )