Coverage for little_loops / recursive_finalize.py: 0%
93 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -0500
1"""Decomposed-parent lifecycle + EPIC re-linking for the rn-implement loops.
3When ``rn-decompose`` splits a parent issue into children, the parent issue must
4be closed (work is now carried by the children) and, if the parent belonged to an
5EPIC, its children must be re-linked into that EPIC so they do not silently fall
6out of the EPIC's ``relates_to:``/``## Children`` rollup (ENH-1977, GAP F).
8This module is the tested core behind ``ll-issues finalize-decomposition``. It is
9deliberately filesystem-only (no git, no Logger, no BRConfig) so it can be unit
10tested against a temporary ``.issues`` tree.
12Field-collision decision (ENH-1977 Fix 4): ``parent:`` is canonically used for
13*both* decomposition lineage and EPIC membership. A child cannot carry two
14``parent:`` values, so children become first-class EPIC members
15(``parent: EPIC-NNN``) and decomposition lineage is recorded via
16``relates_to: [<parent-id>]`` plus the ``Decomposed from <parent-id>`` body marker
17that ``issue-size-review`` already emits.
18"""
20from __future__ import annotations
22import re
23import shutil
24import subprocess
25from datetime import UTC, datetime
26from pathlib import Path
27from typing import Any
29from little_loops.frontmatter import parse_frontmatter, update_frontmatter
31_EPIC_RE = re.compile(r"^EPIC-\d+$", re.IGNORECASE)
34def _completed_at_now() -> str:
35 """Return current UTC time as ISO 8601 with a ``Z`` suffix."""
36 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
39def _find_issue_file(
40 issue_id: str, issues_dir: Path, *, include_completed: bool = True
41) -> Path | None:
42 """Locate the markdown file for ``issue_id`` under ``issues_dir``.
44 Matches files whose name contains ``-<issue_id>-`` (case-insensitive), mirroring
45 the ``*-$ID-*`` glob used by the loops. Active category directories are searched
46 first; the completed directory is only searched when ``include_completed`` is set.
47 """
48 needle = f"-{issue_id.upper()}-"
49 candidates: list[Path] = []
50 for path in sorted(issues_dir.rglob("*.md")):
51 if not include_completed and "completed" in path.parts:
52 continue
53 if needle in f"-{path.name.upper()}":
54 # Prefer active files over completed ones when both exist.
55 candidates.append(path)
56 if not candidates:
57 return None
58 active = [p for p in candidates if "completed" not in p.parts]
59 return active[0] if active else candidates[0]
62def _dedup(seq: list[str]) -> list[str]:
63 """Order-preserving de-duplication (case-insensitive keys, original casing kept)."""
64 seen: set[str] = set()
65 out: list[str] = []
66 for item in seq:
67 key = item.upper()
68 if key not in seen:
69 seen.add(key)
70 out.append(item)
71 return out
74def _git_mv(src: Path, dst: Path) -> None:
75 """Move ``src`` to ``dst`` using ``git mv`` when possible, else a plain move."""
76 dst.parent.mkdir(parents=True, exist_ok=True)
77 try:
78 result = subprocess.run(
79 ["git", "mv", str(src), str(dst)],
80 capture_output=True,
81 text=True,
82 cwd=src.parent,
83 )
84 if result.returncode == 0:
85 return
86 except (OSError, subprocess.SubprocessError):
87 pass
88 shutil.move(str(src), str(dst))
91def _append_decomposition_note(content: str, parent_id: str, child_ids: list[str]) -> str:
92 """Append a ``## Resolution`` decomposition note if not already present.
94 Idempotent: a second call is a no-op once the marker line exists.
95 """
96 if "Decomposed into" in content:
97 return content
98 children_str = ", ".join(child_ids) if child_ids else "(no children recorded)"
99 note = (
100 "\n\n---\n\n## Resolution\n\n"
101 f"- **Status**: Decomposed\n"
102 f"- **Closed**: {datetime.now(UTC).strftime('%Y-%m-%d')}\n"
103 f"- **Decomposed into**: {children_str}\n\n"
104 f"Work for {parent_id} is now carried by its child issues; this parent was "
105 f"closed by rn-decompose.\n"
106 )
107 return content.rstrip() + note
110def finalize_decomposed_parent(
111 parent_id: str,
112 child_ids: list[str],
113 issues_dir: Path,
114 *,
115 move_to_completed: bool = True,
116) -> dict[str, Any]:
117 """Close a decomposed parent and re-link its children into the parent's EPIC.
119 Idempotent across all steps so a retried loop iteration does not corrupt state.
121 Steps:
122 1. **Close the parent.** Set ``status: done`` + ``completed_at``, append a
123 "Decomposed into <child-ids>" body note, and (when ``move_to_completed``)
124 move the file into ``<issues_dir>/completed/``.
125 2. **Re-link children to the parent's EPIC, if any.** When the parent carries
126 ``parent: EPIC-NNN``: repoint each child to ``parent: EPIC-NNN``, record
127 lineage via ``relates_to: [<parent-id>]``, append the children to the
128 EPIC's ``relates_to:``, and drop the decomposed parent from it.
129 3. **No-EPIC guard.** When the parent has no EPIC parent, only step 1 runs and
130 children keep their existing ``parent:`` linkage.
132 Args:
133 parent_id: Bare issue ID of the decomposed parent (e.g. ``ENH-123``).
134 child_ids: Child issue IDs created from the parent.
135 issues_dir: Root of the issues tree (e.g. ``Path(".issues")``).
136 move_to_completed: Move the closed parent into ``completed/`` when true.
138 Returns:
139 Summary dict: ``{"parent", "epic", "children", "moved", "warnings"}``.
140 """
141 parent_id = parent_id.strip().upper()
142 child_ids = [c.strip() for c in child_ids if c.strip()]
143 warnings: list[str] = []
145 parent_path = _find_issue_file(parent_id, issues_dir)
146 if parent_path is None:
147 return {
148 "parent": parent_id,
149 "epic": None,
150 "children": child_ids,
151 "moved": False,
152 "warnings": [f"parent file not found for {parent_id}"],
153 }
155 content = parent_path.read_text(encoding="utf-8")
156 fm = parse_frontmatter(content)
157 raw_epic = fm.get("parent")
158 epic_id: str | None = None
159 if isinstance(raw_epic, str) and _EPIC_RE.match(raw_epic.strip()):
160 epic_id = raw_epic.strip().upper()
162 # --- Step 1: close the parent -------------------------------------------------
163 content = _append_decomposition_note(content, parent_id, child_ids)
164 content = update_frontmatter(content, {"status": "done", "completed_at": _completed_at_now()})
165 parent_path.write_text(content, encoding="utf-8")
167 moved = False
168 if move_to_completed and "completed" not in parent_path.parts:
169 completed_dir = issues_dir / "completed"
170 dst = completed_dir / parent_path.name
171 if not dst.exists():
172 _git_mv(parent_path, dst)
173 moved = True
175 # --- Steps 2/3: EPIC re-linking (guarded) -------------------------------------
176 if epic_id is not None:
177 for child_id in child_ids:
178 child_path = _find_issue_file(child_id, issues_dir)
179 if child_path is None:
180 warnings.append(f"child file not found for {child_id}")
181 continue
182 child_content = child_path.read_text(encoding="utf-8")
183 child_fm = parse_frontmatter(child_content)
184 existing_relates = list(child_fm.get("relates_to") or [])
185 new_relates = _dedup([*existing_relates, parent_id])
186 child_content = update_frontmatter(
187 child_content, {"parent": epic_id, "relates_to": new_relates}
188 )
189 child_path.write_text(child_content, encoding="utf-8")
191 epic_path = _find_issue_file(epic_id, issues_dir)
192 if epic_path is None:
193 warnings.append(f"epic file not found for {epic_id}")
194 else:
195 epic_content = epic_path.read_text(encoding="utf-8")
196 epic_fm = parse_frontmatter(epic_content)
197 existing = [str(x) for x in (epic_fm.get("relates_to") or [])]
198 # Add children, drop the now-decomposed parent.
199 merged = _dedup([*existing, *child_ids])
200 merged = [x for x in merged if x.upper() != parent_id]
201 epic_content = update_frontmatter(epic_content, {"relates_to": merged})
202 epic_path.write_text(epic_content, encoding="utf-8")
204 return {
205 "parent": parent_id,
206 "epic": epic_id,
207 "children": child_ids,
208 "moved": moved,
209 "warnings": warnings,
210 }