Coverage for little_loops / cli / issues / epic_consistency.py: 0%
172 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""ll-issues epic-consistency: detect and reconcile EPIC body/parent drift (FEAT-2332)."""
3from __future__ import annotations
5import argparse
6import re
7from dataclasses import dataclass, field
8from pathlib import Path
9from typing import TYPE_CHECKING
11if TYPE_CHECKING:
12 from little_loops.config import BRConfig
13 from little_loops.issue_parser import IssueInfo
15_ALL_STATUSES: set[str] = {"open", "in_progress", "blocked", "done", "cancelled", "deferred"}
17# Matches the frontmatter type: field value (captures the value after "type: ")
18_FM_TYPE_RE = re.compile(r"^type:\s*(\S+)", re.MULTILINE)
19# Matches a frontmatter children: key (list start)
20_FM_CHILDREN_RE = re.compile(r"^children\s*:", re.MULTILINE)
22# Issue types that participate in the (a)/(b) parent-child diff.
23# EPIC-* refs in body are treated as sub-epic prose (advisory only).
24# Everything else (MR-*, CT-*, EG-*, …) is skipped.
25_REAL_ISSUE_TYPES = frozenset({"BUG", "FEAT", "ENH"})
27# Matches bullet list items with an optional bold wrapper around the ID:
28# - FEAT-001 … → group(1) = "FEAT-001"
29# - **FEAT-001** — … → group(1) = "FEAT-001"
30_BODY_BULLET_RE = re.compile(r"^\s*[-*]\s+\*{0,2}([A-Z]+-\d+)", re.MULTILINE)
33@dataclass
34class EpicDrift:
35 """Drift report for a single EPIC."""
37 epic_id: str
38 epic_title: str
39 missing_from_body: list[str] = field(default_factory=list) # (a)
40 body_without_parent: list[str] = field(default_factory=list) # (b)
41 relates_to_is_child: list[str] = field(default_factory=list) # (c)
42 sub_epic_advisory: list[str] = field(default_factory=list) # advisory
43 type_casing_wrong: bool = False # schema: type: must be 'EPIC'
44 has_children_frontmatter: bool = False # schema: children: key forbidden
46 @property
47 def has_drift(self) -> bool:
48 """True when any actionable drift exists (a/b/c categories or schema violations)."""
49 return bool(
50 self.missing_from_body
51 or self.body_without_parent
52 or self.relates_to_is_child
53 or self.type_casing_wrong
54 or self.has_children_frontmatter
55 )
57 def to_dict(self) -> dict:
58 """Serialize to a JSON-serializable dict for --format json output."""
59 return {
60 "epic": self.epic_id,
61 "missing_from_body": sorted(self.missing_from_body),
62 "body_without_parent": sorted(self.body_without_parent),
63 "relates_to_is_child": sorted(self.relates_to_is_child),
64 "sub_epic_advisory": sorted(self.sub_epic_advisory),
65 "type_casing_wrong": self.type_casing_wrong,
66 "has_children_frontmatter": self.has_children_frontmatter,
67 }
70def _section_bounds(content: str, heading: str) -> tuple[int, int] | None:
71 """Return (body_start, body_end) byte offsets for a ## heading section.
73 Returns None when the heading is absent.
74 """
75 pattern = rf"^##\s+{re.escape(heading)}\s*$"
76 match = re.search(pattern, content, re.MULTILINE)
77 if not match:
78 return None
79 start = match.end()
80 next_match = re.search(r"^##\s", content[start:], re.MULTILINE)
81 end = start + next_match.start() if next_match else len(content)
82 return start, end
85def _parse_children_body(section_text: str) -> tuple[set[str], set[str]]:
86 """Parse a ## Children section body.
88 Returns:
89 (real_issue_ids, sub_epic_ids) where real_issue_ids contains only
90 BUG/FEAT/ENH tokens and sub_epic_ids contains EPIC-* tokens.
91 Non-issue tokens (MR-*, CT-*, EG-*, …) are silently dropped.
92 """
93 real_ids: set[str] = set()
94 sub_epic_ids: set[str] = set()
95 for m in _BODY_BULLET_RE.finditer(section_text):
96 token = m.group(1)
97 issue_type = token.split("-")[0]
98 if issue_type in _REAL_ISSUE_TYPES:
99 real_ids.add(token)
100 elif issue_type == "EPIC":
101 sub_epic_ids.add(token)
102 # else: skip non-issue tokens (MR-*, CT-*, EG-*, …)
103 return real_ids, sub_epic_ids
106def compute_drift(
107 epic_id: str,
108 all_issues: list[IssueInfo],
109) -> EpicDrift | None:
110 """Compute drift for a single EPIC against all loaded issues.
112 Returns None when the EPIC ID is not found in all_issues.
113 """
114 epic_id = epic_id.upper()
116 epic_matches = [i for i in all_issues if i.issue_id == epic_id]
117 if not epic_matches:
118 return None
119 epic_info = epic_matches[0]
121 # Authoritative child set: issues carrying parent: epic_id (real types only)
122 parent_children: set[str] = {
123 i.issue_id
124 for i in all_issues
125 if i.parent == epic_id and i.issue_id.split("-")[0] in _REAL_ISSUE_TYPES
126 }
128 # Advisory: EPICs that carry parent: epic_id (should use relates_to + prose)
129 sub_epic_advisory: list[str] = sorted(
130 i.issue_id for i in all_issues if i.parent == epic_id and i.issue_id.startswith("EPIC-")
131 )
133 # Parse file content for body and frontmatter checks
134 try:
135 content = epic_info.path.read_text(encoding="utf-8")
136 except OSError:
137 content = ""
139 # Extract frontmatter block (between first pair of --- delimiters)
140 fm_block = ""
141 if content.startswith("---"):
142 end_marker = content.find("\n---", 3)
143 if end_marker != -1:
144 fm_block = content[3:end_marker]
146 # Schema check (d): type: must be 'EPIC' when present; absent is OK
147 type_casing_wrong = False
148 type_match = _FM_TYPE_RE.search(fm_block)
149 if type_match and type_match.group(1) != "EPIC":
150 type_casing_wrong = True
152 # Schema check (e): children: frontmatter key is forbidden
153 has_children_frontmatter = bool(_FM_CHILDREN_RE.search(fm_block))
155 bounds = _section_bounds(content, "Children")
156 if bounds is not None:
157 section_body = content[bounds[0] : bounds[1]]
158 body_real_ids, _ = _parse_children_body(section_body)
159 else:
160 body_real_ids = set()
162 # (a) parent: child missing from body
163 missing_from_body = sorted(parent_children - body_real_ids)
165 # (b) body-listed real issue with no parent: backref
166 body_without_parent = sorted(body_real_ids - parent_children)
168 # (c) relates_to: entries that are also parent: children
169 relates_to_set = set(epic_info.relates_to or [])
170 relates_to_is_child = sorted(relates_to_set & parent_children)
172 return EpicDrift(
173 epic_id=epic_id,
174 epic_title=epic_info.title,
175 missing_from_body=missing_from_body,
176 body_without_parent=body_without_parent,
177 relates_to_is_child=relates_to_is_child,
178 sub_epic_advisory=sub_epic_advisory,
179 type_casing_wrong=type_casing_wrong,
180 has_children_frontmatter=has_children_frontmatter,
181 )
184def fix_epic(epic_path: Path, missing_from_body: list[str]) -> None:
185 """Add missing parent: children to an EPIC's ## Children section.
187 Existing lines (sub-epic prose, non-issue tokens, category-(b) entries)
188 are preserved. Only category-(a) drift is fixed. Running twice on an
189 already-fixed file is a no-op (idempotent).
190 """
191 if not missing_from_body:
192 return
194 content = epic_path.read_text(encoding="utf-8")
195 new_bullets = "\n".join(
196 f"- **{child_id}** — (added by epic-consistency --fix)"
197 for child_id in sorted(missing_from_body)
198 )
200 bounds = _section_bounds(content, "Children")
201 if bounds is None:
202 # No ## Children section — create one at the end of the file
203 new_section = "\n## Children\n\n" + new_bullets + "\n"
204 new_content = content.rstrip("\n") + "\n" + new_section
205 else:
206 start, end = bounds
207 section_body = content[start:end]
208 # Append after existing content (preserving all existing lines)
209 stripped = section_body.rstrip("\n")
210 sep = "\n" if stripped.strip() else ""
211 new_section_body = stripped + sep + "\n" + new_bullets + "\n"
212 new_content = content[:start] + new_section_body + content[end:]
214 from little_loops.file_utils import atomic_write
216 atomic_write(epic_path, new_content)
219def add_epic_consistency_parser(subs: argparse._SubParsersAction) -> argparse.ArgumentParser:
220 """Register the epic-consistency subparser on *subs*."""
221 from little_loops.cli_args import add_config_arg
223 p = subs.add_parser(
224 "epic-consistency",
225 aliases=["ec"],
226 help="Detect and reconcile EPIC body/parent drift",
227 )
228 p.set_defaults(command="epic-consistency")
229 p.add_argument(
230 "epic_id",
231 nargs="?",
232 default=None,
233 help="EPIC ID (e.g., EPIC-1773); omit when using --all",
234 )
235 p.add_argument("--all", "-a", action="store_true", help="Check all EPICs in epics dir")
236 p.add_argument(
237 "--fix",
238 action="store_true",
239 help="Rewrite ## Children for category-(a) drift (report-only by default)",
240 )
241 p.add_argument(
242 "--format",
243 "-f",
244 choices=["text", "json"],
245 default="text",
246 help="Output format (default: text)",
247 )
248 add_config_arg(p)
249 return p
252def cmd_epic_consistency(config: BRConfig, args: argparse.Namespace) -> int:
253 """Detect and reconcile EPIC body/parent drift.
255 Returns:
256 0 when no drift found (or --fix ran cleanly), 1 on drift/error.
257 """
258 from little_loops.cli.output import print_json
259 from little_loops.issue_parser import find_issues
261 epic_id: str | None = getattr(args, "epic_id", None)
262 check_all: bool = getattr(args, "all", False)
263 fix: bool = getattr(args, "fix", False)
264 fmt: str = getattr(args, "format", "text") or "text"
266 if not epic_id and not check_all:
267 print("Error: provide an EPIC-ID or --all")
268 return 1
270 if epic_id and not epic_id.upper().startswith("EPIC-"):
271 print(f"Error: expected an EPIC ID (e.g., EPIC-1773), got {epic_id!r}")
272 return 1
274 all_issues = find_issues(config, status_filter=_ALL_STATUSES)
276 if check_all:
277 epic_ids = sorted(i.issue_id for i in all_issues if i.issue_id.startswith("EPIC-"))
278 else:
279 epic_ids = [epic_id.upper()] # type: ignore[union-attr]
281 results: list[EpicDrift] = []
282 for eid in epic_ids:
283 drift = compute_drift(eid, all_issues)
284 if drift is None:
285 print(f"Error: EPIC {eid!r} not found")
286 return 1
287 results.append(drift)
289 if fix:
290 for drift in results:
291 if drift.missing_from_body:
292 epic_issues = [i for i in all_issues if i.issue_id == drift.epic_id]
293 if epic_issues:
294 fix_epic(epic_issues[0].path, drift.missing_from_body)
295 # Re-check after fix so output reflects the updated state
296 results = []
297 for eid in epic_ids:
298 drift = compute_drift(eid, all_issues)
299 if drift is not None:
300 results.append(drift)
302 any_drift = any(d.has_drift for d in results)
304 if fmt == "json":
305 if len(results) == 1:
306 print_json(results[0].to_dict())
307 else:
308 print_json({"results": [d.to_dict() for d in results]})
309 return 0 if fix else (1 if any_drift else 0)
311 # text format
312 for drift in results:
313 if not drift.has_drift and not drift.sub_epic_advisory:
314 print(f"{drift.epic_id}: {drift.epic_title} — OK")
315 continue
316 print(f"{drift.epic_id}: {drift.epic_title}")
317 if drift.type_casing_wrong:
318 print(" (d) Schema: type: casing must be 'EPIC' (not lowercase 'epic')")
319 if drift.has_children_frontmatter:
320 print(" (e) Schema: frontmatter children: array is forbidden; use parent: backrefs")
321 if drift.missing_from_body:
322 print(" (a) Missing from body (parent: child not documented):")
323 for child_id in drift.missing_from_body:
324 print(f" {child_id}")
325 if drift.body_without_parent:
326 print(" (b) Body-listed, no parent: backref (human decision needed):")
327 for child_id in drift.body_without_parent:
328 print(f" {child_id}")
329 if drift.relates_to_is_child:
330 print(" (c) relates_to: leaking membership:")
331 for child_id in drift.relates_to_is_child:
332 print(f" {child_id}")
333 if drift.sub_epic_advisory:
334 print(" [advisory] Sub-epic via parent: (should use relates_to + prose):")
335 for child_id in drift.sub_epic_advisory:
336 print(f" {child_id}")
338 return 0 if fix else (1 if any_drift else 0)