Coverage for little_loops / cli / issues / clusters.py: 0%
229 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""ll-issues clusters: Render issue relationship clusters as box diagrams."""
3from __future__ import annotations
5import argparse
6from collections import deque
7from typing import TYPE_CHECKING
9from little_loops.cli.output import colorize, print_json
11if TYPE_CHECKING:
12 from little_loops.config import BRConfig
13 from little_loops.issue_parser import IssueInfo
15# ANSI color codes per relationship type
16EDGE_COLOR: dict[str, str] = {
17 "blocks": "31", # red
18 "blocked_by": "33", # yellow
19 "parent": "34", # blue
20 "sibling": "36", # cyan
21 "depends_on": "35", # magenta
22 "relates_to": "37", # white/dim
23}
25_BOX_HEIGHT = 4 # top border + 2 content lines + bottom border
26_GAP_HEIGHT = 2 # rows between boxes for arrow drawing
27_BOX_MARGIN = 2 # left-margin column offset
28_MAX_BOX_WIDTH = 60
30# Edge type sets for --edges aliases
31_ALL_EDGE_TYPES = frozenset({"blocked_by", "blocks", "depends_on", "relates_to", "parent"})
32_BLOCKING_EDGE_TYPES = frozenset({"blocked_by", "blocks"})
33_HARD_EDGE_TYPES = frozenset({"blocked_by", "blocks", "depends_on"})
35# Active status set for --status=active default
36_ACTIVE_STATUSES = frozenset({"open", "in_progress", "blocked"})
38# Priority order when two relationships describe the same pair (lower = higher priority)
39_EDGE_PRIORITY: dict[str, int] = {
40 "blocked_by": 0,
41 "blocks": 1,
42 "parent": 2,
43 "depends_on": 3,
44 "relates_to": 4,
45}
48def _resolve_edge_types(edges_arg: str) -> set[str]:
49 """Resolve --edges argument to a set of edge type strings."""
50 if edges_arg == "all":
51 return set(_ALL_EDGE_TYPES)
52 if edges_arg == "blocking":
53 return set(_BLOCKING_EDGE_TYPES)
54 if edges_arg == "hard":
55 return set(_HARD_EDGE_TYPES)
56 return set(edges_arg.split(","))
59def _resolve_status_set(status_arg: str) -> set[str]:
60 """Resolve --status argument to a set of canonical status strings."""
61 if status_arg == "active":
62 return set(_ACTIVE_STATUSES)
63 if status_arg == "+deferred":
64 return set(_ACTIVE_STATUSES) | {"deferred"}
65 if status_arg == "all":
66 return {"open", "in_progress", "blocked", "done", "deferred"}
67 return set(status_arg.split(","))
70def _build_neighbour_map(issues: list[IssueInfo], edge_types: set[str]) -> dict[str, set[str]]:
71 """Build undirected neighbour map from IssueInfo for connectivity BFS.
73 Only connects issues to other issues present in the loaded list.
74 """
75 issue_ids = {i.issue_id for i in issues}
76 neighbours: dict[str, set[str]] = {i.issue_id: set() for i in issues}
78 for issue in issues:
79 iid = issue.issue_id
80 targets: list[str] = []
82 if "blocked_by" in edge_types:
83 targets.extend(issue.blocked_by)
84 if "blocks" in edge_types:
85 targets.extend(issue.blocks)
86 if "depends_on" in edge_types:
87 targets.extend(issue.depends_on)
88 if "relates_to" in edge_types:
89 targets.extend(issue.relates_to)
90 if "parent" in edge_types and issue.parent:
91 targets.append(issue.parent)
93 for target in targets:
94 if target in issue_ids:
95 neighbours[iid].add(target)
96 neighbours[target].add(iid)
98 return neighbours
101def _get_components(neighbours: dict[str, set[str]]) -> list[list[str]]:
102 """BFS over undirected neighbour map to find connected components.
104 Returns components sorted by size descending.
105 """
106 visited: set[str] = set()
107 components: list[list[str]] = []
109 for node in sorted(neighbours):
110 if node in visited:
111 continue
112 component: list[str] = []
113 queue: deque[str] = deque([node])
114 while queue:
115 current = queue.popleft()
116 if current in visited:
117 continue
118 visited.add(current)
119 component.append(current)
120 for neighbor in sorted(neighbours.get(current, set())):
121 if neighbor not in visited:
122 queue.append(neighbor)
123 components.append(component)
125 return sorted(components, key=len, reverse=True)
128def _topo_sort_cluster(
129 cluster_ids: list[str],
130 blocked_by: dict[str, set[str]],
131) -> tuple[list[str], bool]:
132 """Topological sort (Kahn's) scoped to this cluster.
134 Returns (sorted_ids, has_cycle). Nodes in cycles are appended in sorted
135 order after the acyclic prefix so the caller always gets a full list.
136 """
137 cluster_set = set(cluster_ids)
138 in_degree: dict[str, int] = dict.fromkeys(cluster_ids, 0)
139 adj: dict[str, list[str]] = {id_: [] for id_ in cluster_ids}
141 for id_ in cluster_ids:
142 for dep in sorted(blocked_by.get(id_, set()) & cluster_set):
143 in_degree[id_] += 1
144 adj[dep].append(id_)
146 queue: deque[str] = deque(sorted(id_ for id_, deg in in_degree.items() if deg == 0))
147 result: list[str] = []
149 while queue:
150 node = queue.popleft()
151 result.append(node)
152 for dep in sorted(adj[node]):
153 in_degree[dep] -= 1
154 if in_degree[dep] == 0:
155 queue.append(dep)
157 has_cycle = len(result) < len(cluster_ids)
158 if has_cycle:
159 remaining = sorted(id_ for id_ in cluster_ids if id_ not in set(result))
160 result.extend(remaining)
162 return result, has_cycle
165def _cluster_edges(
166 cluster_ids: set[str],
167 issues: list[IssueInfo],
168 edge_types: set[str],
169) -> list[tuple[str, str, str]]:
170 """Return directed edges within a cluster as (from_id, to_id, relationship).
172 Deduplicates edges between the same pair, keeping the highest-priority
173 relationship type (blocked_by > blocks > parent > depends_on > relates_to).
174 """
175 issues_in_cluster = {i.issue_id: i for i in issues if i.issue_id in cluster_ids}
176 # frozenset key → (from_id, to_id, rel) for deduplication
177 best: dict[frozenset[str], tuple[str, str, str]] = {}
179 for iid in sorted(cluster_ids):
180 issue = issues_in_cluster.get(iid)
181 if not issue:
182 continue
184 candidates: list[tuple[str, str, str]] = []
185 if "blocked_by" in edge_types:
186 for t in sorted(issue.blocked_by):
187 if t in cluster_ids:
188 candidates.append((iid, t, "blocked_by"))
189 if "blocks" in edge_types:
190 for t in sorted(issue.blocks):
191 if t in cluster_ids:
192 candidates.append((iid, t, "blocks"))
193 if "depends_on" in edge_types:
194 for t in sorted(issue.depends_on):
195 if t in cluster_ids:
196 candidates.append((iid, t, "depends_on"))
197 if "relates_to" in edge_types:
198 for t in sorted(issue.relates_to):
199 if t in cluster_ids:
200 candidates.append((iid, t, "relates_to"))
201 if "parent" in edge_types and issue.parent and issue.parent in cluster_ids:
202 candidates.append((iid, issue.parent, "parent"))
204 for from_id, to_id, rel in candidates:
205 key: frozenset[str] = frozenset({from_id, to_id})
206 existing = best.get(key)
207 if existing is None or _EDGE_PRIORITY.get(rel, 99) < _EDGE_PRIORITY.get(
208 existing[2], 99
209 ):
210 best[key] = (from_id, to_id, rel)
212 return list(best.values())
215def _render_cluster_diagram(
216 ordered_ids: list[str],
217 issues_map: dict[str, IssueInfo],
218 edge_map: dict[tuple[str, str], str],
219 box_width: int,
220) -> list[str]:
221 """Render a cluster as a vertical stack of box diagrams with arrows.
223 Uses _draw_box from cli/loop/layout.py as the box primitive.
224 Arrows are drawn between consecutive nodes; edge labels are appended
225 after the arrow row so ANSI escapes don't corrupt the character grid.
226 """
227 # Intentional cross-module import of a private primitive; _draw_box is
228 # reusable and has no FSM-specific logic when is_highlighted=False.
229 from little_loops.cli.loop.layout import _draw_box
231 n = len(ordered_ids)
232 grid_h = n * _BOX_HEIGHT + max(0, n - 1) * _GAP_HEIGHT
233 grid_w = box_width + _BOX_MARGIN * 2 + 2
234 grid: list[list[str]] = [[" "] * grid_w for _ in range(grid_h)]
236 center_col = _BOX_MARGIN + box_width // 2
237 box_start: dict[str, int] = {}
238 avail = box_width - 4 # interior width minus side margins
240 for i, issue_id in enumerate(ordered_ids):
241 row = i * (_BOX_HEIGHT + _GAP_HEIGHT)
242 issue = issues_map[issue_id]
243 title = issue.title if len(issue.title) <= avail else issue.title[: avail - 1] + "…"
244 content = [f"[{issue.priority}] {issue_id}", title]
245 _draw_box(grid, row, _BOX_MARGIN, box_width, _BOX_HEIGHT, content, False, "0")
246 box_start[issue_id] = row
248 # Annotate gap rows with arrow characters and colored edge labels.
249 # Only draw connectors when a real edge exists between consecutive nodes.
250 arrow_labels: dict[int, str] = {}
251 for i in range(n - 1):
252 a_id = ordered_ids[i]
253 b_id = ordered_ids[i + 1]
254 gap_row = box_start[a_id] + _BOX_HEIGHT
256 rel = edge_map.get((a_id, b_id)) or edge_map.get((b_id, a_id))
257 if rel:
258 if gap_row < grid_h:
259 grid[gap_row][center_col] = "│"
260 if gap_row + 1 < grid_h:
261 grid[gap_row + 1][center_col] = "▼"
262 color = EDGE_COLOR.get(rel, "37")
263 arrow_labels[gap_row] = colorize(f" {rel}", color)
265 # Convert grid to string lines, appending annotations after arrow rows
266 lines: list[str] = []
267 for r, row_chars in enumerate(grid):
268 line = "".join(row_chars).rstrip()
269 if r in arrow_labels:
270 line += arrow_labels[r]
271 lines.append(line)
273 while lines and not lines[-1].strip():
274 lines.pop()
276 # Append annotations for skip-level edges (non-consecutive in topo order).
277 pos = {id_: i for i, id_ in enumerate(ordered_ids)}
278 skip_edges = [(f, t, r) for (f, t), r in sorted(edge_map.items()) if abs(pos[t] - pos[f]) > 1]
279 if skip_edges:
280 lines.append("")
281 for src, dst, rel in skip_edges:
282 color = EDGE_COLOR.get(rel, "37")
283 lines.append(f" {src} {colorize('→', color)} {dst} ({rel})")
285 return lines
288def cmd_clusters(config: BRConfig, args: argparse.Namespace) -> int:
289 """Render issue relationship clusters as box diagrams.
291 Args:
292 config: Project configuration (provides issue directories and CLI settings)
293 args: Parsed CLI args (include_orphans, min_connections, json, edges, status)
295 Returns:
296 Exit code (0 = success)
297 """
298 from little_loops.cli.output import terminal_width
299 from little_loops.issue_parser import find_issues
301 edges_arg: str = getattr(args, "edges", "all")
302 status_arg: str = getattr(args, "status", "active")
304 edge_types = _resolve_edge_types(edges_arg)
305 status_set = _resolve_status_set(status_arg)
307 issues = find_issues(config, status_filter=status_set)
308 if not issues:
309 print("No active issues found.")
310 return 0
312 neighbours = _build_neighbour_map(issues, edge_types)
313 issues_map = {issue.issue_id: issue for issue in issues}
314 components = _get_components(neighbours)
316 include_orphans: bool = getattr(args, "include_orphans", False)
317 if not include_orphans:
318 components = [c for c in components if len(c) > 1]
320 min_conn: int = getattr(args, "min_connections", 0) or 0
321 if min_conn > 0:
323 def _max_degree(comp: list[str]) -> int:
324 return max(len(neighbours.get(id_, set())) for id_ in comp)
326 components = [c for c in components if _max_degree(c) >= min_conn]
328 if not components:
329 all_components = _get_components(neighbours)
330 if all(len(c) == 1 for c in all_components):
331 print("No issue relationships found. Use --include-orphans to show isolated issues.")
332 else:
333 print("No clusters match the specified filters.")
334 return 0
336 # JSON mode: emit structured data, no diagram rendering
337 if getattr(args, "json", False):
338 output = []
339 for idx, comp in enumerate(components, 1):
340 comp_set = set(comp)
341 edges = _cluster_edges(comp_set, issues, edge_types)
342 output.append(
343 {
344 "cluster_index": idx,
345 "issue_count": len(comp),
346 "issues": [
347 {
348 "id": id_,
349 "priority": issues_map[id_].priority,
350 "title": issues_map[id_].title,
351 }
352 for id_ in sorted(comp)
353 ],
354 "edges": [{"from": f, "to": t, "relationship": r} for f, t, r in edges],
355 }
356 )
357 print_json(output)
358 return 0
360 # Text rendering
361 width = terminal_width()
362 box_w = max(20, min(_MAX_BOX_WIDTH, width - _BOX_MARGIN * 2 - 4))
363 total_issues = sum(len(c) for c in components)
365 # Build blocked_by map from IssueInfo for topo sort ordering
366 blocked_by_map: dict[str, set[str]] = {
367 issue.issue_id: set(issue.blocked_by) for issue in issues
368 }
370 for idx, comp in enumerate(components, 1):
371 comp_set = set(comp)
372 edges = _cluster_edges(comp_set, issues, edge_types)
373 edge_map: dict[tuple[str, str], str] = {(f, t): r for f, t, r in edges}
375 ordered, has_cycle = _topo_sort_cluster(comp, blocked_by_map)
377 sep = "─" * 3
378 n_issues = len(comp)
379 noun = "issue" if n_issues == 1 else "issues"
380 print(f"{sep} Cluster {idx} ({n_issues} {noun}) {sep}")
381 if has_cycle:
382 print("⚠ cycle detected — using fallback layout")
384 diagram_lines = _render_cluster_diagram(ordered, issues_map, edge_map, box_w)
385 print("\n".join(diagram_lines))
386 print()
388 n_clusters = len(components)
389 cluster_noun = "cluster" if n_clusters == 1 else "clusters"
390 issue_noun = "issue" if total_issues == 1 else "issues"
391 print(f"{n_clusters} {cluster_noun}, {total_issues} {issue_noun} total")
392 return 0