Coverage for little_loops / cli / sprint / show.py: 8%
142 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-03-18 16:18 -0500
1"""ll-sprint show subcommand and dependency visualization renderers."""
3from __future__ import annotations
5import argparse
6from pathlib import Path
7from typing import TYPE_CHECKING, Any
9from little_loops.cli.output import colorize, terminal_width
10from little_loops.cli.sprint._helpers import (
11 _build_issue_contents,
12 _render_dependency_analysis,
13 _render_execution_plan,
14)
15from little_loops.dependency_graph import DependencyGraph, refine_waves_for_contention
16from little_loops.logger import Logger
18if TYPE_CHECKING:
19 from little_loops.dependency_graph import WaveContentionNote
20 from little_loops.sprint import SprintManager
23def _render_dependency_graph(
24 waves: list[list[Any]],
25 dep_graph: DependencyGraph,
26) -> str:
27 """Render ASCII dependency graph.
29 Args:
30 waves: List of execution waves
31 dep_graph: DependencyGraph for looking up relationships
33 Returns:
34 Formatted string showing dependency arrows
35 """
36 if not waves or len(waves) <= 1:
37 return ""
39 # Don't render graph if there are no actual dependency edges
40 # (waves > 1 can happen from file overlap splitting alone)
41 all_ids = {issue.issue_id for wave in waves for issue in wave}
42 has_edges = any(dep_graph.blocks.get(issue_id, set()) & all_ids for issue_id in all_ids)
43 if not has_edges:
44 return ""
46 width = terminal_width()
47 lines: list[str] = []
48 lines.append("")
49 lines.append("=" * width)
50 lines.append("DEPENDENCY GRAPH")
51 lines.append("=" * width)
52 lines.append("")
54 # Build chains: track which issues block what
55 # Show each independent chain on its own line
56 chains: list[str] = []
57 visited: set[str] = set()
59 def build_chain(issue_id: str) -> str:
60 """Recursively build chain string from issue."""
61 if issue_id in visited:
62 return issue_id
63 visited.add(issue_id)
65 blocked_issues = sorted(dep_graph.blocks.get(issue_id, set()))
66 if not blocked_issues:
67 return issue_id
69 if len(blocked_issues) == 1:
70 return f"{issue_id} \u2500\u2500\u2192 {build_chain(blocked_issues[0])}"
71 else:
72 # Multiple branches - show first inline, note others
73 result = f"{issue_id} \u2500\u2500\u2192 {build_chain(blocked_issues[0])}"
74 for other in blocked_issues[1:]:
75 if other not in visited:
76 chains.append(f" {issue_id} \u2500\u2500\u2192 {build_chain(other)}")
77 return result
79 # Find root issues structurally (not blocked by anything in this graph)
80 roots = [iid for iid in sorted(all_ids) if not (dep_graph.blocked_by.get(iid, set()) & all_ids)]
82 for root in roots:
83 if root not in visited:
84 chain = build_chain(root)
85 if chain and "──→" in chain:
86 chains.append(f" {chain}")
88 lines.extend(chains)
89 lines.append("")
90 lines.append("Legend: \u2500\u2500\u2192 blocks (must complete before)")
92 return "\n".join(lines)
95def _render_health_summary(
96 waves: list[list[Any]],
97 contention_notes: list[WaveContentionNote | None] | None,
98 has_cycles: bool,
99 invalid: set[str],
100 dep_report: Any | None = None,
101 issue_to_wave: dict[str, int] | None = None,
102) -> str:
103 """Render a one-line sprint health summary.
105 Returns:
106 Health summary string like "OK -- 5 issues in 1 wave, contention serialized"
107 """
108 total_issues = sum(len(w) for w in waves)
110 _STATUS_COLOR = {"OK": "32", "REVIEW": "33", "WARNING": "38;5;208", "BLOCKED": "31"}
112 if has_cycles:
113 return f"{colorize('BLOCKED', _STATUS_COLOR['BLOCKED'])} -- dependency cycles detected"
115 if invalid:
116 return f"{colorize('WARNING', _STATUS_COLOR['WARNING'])} -- {len(invalid)} issue(s) not found on disk"
118 # Check for novel (unsatisfied) high-confidence proposals
119 if dep_report and dep_report.proposals and issue_to_wave is not None:
120 novel_count = 0
121 for p in dep_report.proposals:
122 target_wave = issue_to_wave.get(p.target_id)
123 source_wave = issue_to_wave.get(p.source_id)
124 if target_wave is None or source_wave is None or target_wave >= source_wave:
125 if p.confidence >= 0.5:
126 novel_count += 1
127 if novel_count > 0:
128 return f"{colorize('REVIEW', _STATUS_COLOR['REVIEW'])} -- {novel_count} potential dependency(ies) to review"
130 # Count logical waves (group contention sub-waves)
131 notes = contention_notes or [None] * len(waves)
132 logical_count = 0
133 has_contention = False
134 prev_parent: int | None = None
135 for idx in range(len(waves)):
136 note = notes[idx] if idx < len(notes) else None
137 if note is not None:
138 has_contention = True
139 if prev_parent is None or note.parent_wave_index != prev_parent:
140 logical_count += 1
141 prev_parent = note.parent_wave_index
142 else:
143 logical_count += 1
144 prev_parent = None
146 wave_word = "wave" if logical_count == 1 else "waves"
147 suffix = ", overlap serialized" if has_contention else ", all parallelizable"
148 if logical_count == 1 and total_issues == 1:
149 suffix = ""
151 return f"{colorize('OK', _STATUS_COLOR['OK'])} -- {total_issues} issues in {logical_count} {wave_word}{suffix}"
154def _cmd_sprint_show(args: argparse.Namespace, manager: SprintManager) -> int:
155 """Show sprint details with dependency visualization."""
156 logger = Logger()
157 sprint = manager.load(args.sprint)
158 if not sprint:
159 logger.error(f"Sprint not found: {args.sprint}")
160 return 1
162 # Validate issues
163 valid = manager.validate_issues(sprint.issues)
164 invalid = set(sprint.issues) - set(valid.keys())
166 # Load full IssueInfo objects for dependency analysis
167 issue_infos = manager.load_issue_infos(list(valid.keys()))
168 dep_graph: DependencyGraph | None = None
169 waves: list[list[Any]] = []
170 contention_notes: list[WaveContentionNote | None] | None = None
171 has_cycles = False
173 # Gather all issue IDs on disk to avoid false "nonexistent" warnings
174 from little_loops.dependency_mapper import gather_all_issue_ids
176 config = manager.config
177 issues_dir = config.project_root / config.issues.base_dir if config else Path(".issues")
178 all_known_ids = gather_all_issue_ids(issues_dir, config=config)
180 if issue_infos:
181 dep_graph = DependencyGraph.from_issues(issue_infos, all_known_ids=all_known_ids)
182 has_cycles = dep_graph.has_cycles()
184 if not has_cycles:
185 waves = dep_graph.get_execution_waves()
186 dep_config = config.dependency_mapping if config else None
187 waves, contention_notes = refine_waves_for_contention(waves, config=dep_config)
189 print(f"{colorize('Sprint:', '1')} {sprint.name}")
190 print(f"Description: {sprint.description or '(none)'}")
191 print(f"Created: {sprint.created}")
193 # Options on a single compact line right after metadata
194 if sprint.options:
195 opts = sprint.options
196 print(
197 f"Options: max_workers={opts.max_workers}, timeout={opts.timeout}s, max_iterations={opts.max_iterations}"
198 )
200 # Dependency analysis (ENH-301) - run before health summary so we can reference it
201 dep_report: Any = None
202 issue_to_wave: dict[str, int] = {}
203 if issue_infos and not args.skip_analysis:
204 from little_loops.dependency_mapper import analyze_dependencies
206 issue_contents = _build_issue_contents(issue_infos)
207 dep_report = analyze_dependencies(
208 issue_infos, issue_contents, all_known_ids=all_known_ids, config=dep_config
209 )
211 # Build wave ordering map so we can filter already-satisfied proposals
212 for wave_idx, wave in enumerate(waves):
213 for issue in wave:
214 issue_to_wave[issue.issue_id] = wave_idx
216 # Sprint health summary
217 if waves:
218 health = _render_health_summary(
219 waves,
220 contention_notes,
221 has_cycles,
222 invalid,
223 dep_report=dep_report,
224 issue_to_wave=issue_to_wave if issue_to_wave else None,
225 )
226 print(f"Sprint health: {health}")
228 # Show execution plan if we have dependency info and no cycles
229 if waves and dep_graph:
230 print(_render_execution_plan(waves, dep_graph, contention_notes))
231 print(_render_dependency_graph(waves, dep_graph))
232 else:
233 # Fallback to simple list if no valid issues or cycles
234 print(f"Issues ({len(sprint.issues)}):")
235 for issue_id in sprint.issues:
236 status = "valid" if issue_id in valid else "NOT FOUND"
237 print(f" - {issue_id} ({status})")
239 # Warn about cycles if detected
240 if has_cycles and dep_graph:
241 cycles = dep_graph.detect_cycles()
242 print("\nWarning: Dependency cycles detected:")
243 for cycle in cycles:
244 print(f" {' -> '.join(cycle)}")
246 # Render dependency analysis output
247 if dep_report is not None:
248 _render_dependency_analysis(
249 dep_report,
250 logger,
251 issue_to_wave=issue_to_wave if issue_to_wave else None,
252 config=dep_config,
253 )
255 if invalid:
256 print(f"\nWarning: {len(invalid)} issue(s) not found")
258 return 0