Coverage for little_loops / cli / sprint / show.py: 7%
216 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:30 -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, format_relative_time, print_json, 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 header_text = "Dependency Graph"
49 fill = "\u2500" * max(0, width - len(header_text) - 4)
50 lines.append("")
51 lines.append(f"\u2500\u2500 {header_text} {fill}")
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 has_unknown_hints = any(n is not None and getattr(n, "has_unknown_hints", False) for n in notes)
148 wave_word = "wave" if logical_count == 1 else "waves"
149 if has_unknown_hints and has_contention:
150 suffix = ", serialized (overlap + missing file hints)"
151 elif has_unknown_hints:
152 suffix = ", serialized (missing file hints)"
153 elif has_contention:
154 suffix = ", overlap serialized"
155 else:
156 suffix = ", all parallelizable"
157 if logical_count == 1 and total_issues == 1:
158 suffix = ""
160 return f"{colorize('OK', _STATUS_COLOR['OK'])} -- {total_issues} issues in {logical_count} {wave_word}{suffix}"
163def _cmd_sprint_show(args: argparse.Namespace, manager: SprintManager) -> int:
164 """Show sprint details with dependency visualization."""
165 logger = Logger()
166 sprint = manager.load_or_resolve(args.sprint)
167 if not sprint:
168 logger.error(f"Sprint not found: {args.sprint}")
169 return 1
171 # Validate issues
172 valid = manager.validate_issues(sprint.issues)
173 invalid = set(sprint.issues) - set(valid.keys())
175 # Load full IssueInfo objects for dependency analysis
176 issue_infos = manager.load_issue_infos(list(valid.keys()))
177 dep_graph: DependencyGraph | None = None
178 waves: list[list[Any]] = []
179 contention_notes: list[WaveContentionNote | None] | None = None
180 has_cycles = False
182 # Gather all issue IDs on disk to avoid false "nonexistent" warnings
183 from little_loops.dependency_mapper import gather_all_issue_ids
185 config = manager.config
186 issues_dir = config.project_root / config.issues.base_dir if config else Path(".issues")
187 all_known_ids = gather_all_issue_ids(issues_dir, config=config)
189 if issue_infos:
190 dep_graph = DependencyGraph.from_issues(issue_infos, all_known_ids=all_known_ids)
191 has_cycles = dep_graph.has_cycles()
193 if not has_cycles:
194 waves = dep_graph.get_execution_waves()
195 dep_config = config.dependency_mapping if config else None
196 waves, contention_notes = refine_waves_for_contention(waves, config=dep_config)
198 # JSON early-exit
199 if getattr(args, "json", False):
200 return _show_json(sprint, issue_infos, waves, contention_notes, has_cycles, dep_graph)
202 print(f"{colorize('Sprint:', '1')} {sprint.name}")
203 if sprint.description:
204 print(f"Description: {sprint.description}")
205 print(f"Created: {_format_created(sprint.created)}")
207 # Options on a single compact line right after metadata
208 if sprint.options:
209 opts = sprint.options
210 print(
211 f"Options: max_workers={opts.max_workers}, timeout={opts.timeout}s, max_iterations={opts.max_iterations}"
212 )
214 # Sprint run state from .sprint-state.json
215 _print_run_state(sprint.name)
217 # Dependency analysis (ENH-301) - run before health summary so we can reference it
218 dep_report: Any = None
219 issue_to_wave: dict[str, int] = {}
220 if issue_infos and not args.skip_analysis:
221 from little_loops.dependency_mapper import analyze_dependencies
223 issue_contents = _build_issue_contents(issue_infos)
224 dep_report = analyze_dependencies(
225 issue_infos, issue_contents, all_known_ids=all_known_ids, config=dep_config
226 )
228 # Build wave ordering map so we can filter already-satisfied proposals
229 for wave_idx, wave in enumerate(waves):
230 for issue in wave:
231 issue_to_wave[issue.issue_id] = wave_idx
233 # Sprint health summary
234 if waves:
235 health = _render_health_summary(
236 waves,
237 contention_notes,
238 has_cycles,
239 invalid,
240 dep_report=dep_report,
241 issue_to_wave=issue_to_wave if issue_to_wave else None,
242 )
243 print(f"Sprint health: {health}")
245 # Composition breakdown
246 _print_composition(issue_infos)
248 # Show execution plan if we have dependency info and no cycles
249 if waves and dep_graph:
250 print(_render_execution_plan(waves, dep_graph, contention_notes, config=dep_config))
251 print(_render_dependency_graph(waves, dep_graph))
252 else:
253 # Fallback to simple list if no valid issues or cycles
254 print(f"Issues ({len(sprint.issues)}):")
255 for issue_id in sprint.issues:
256 status = "valid" if issue_id in valid else "NOT FOUND"
257 print(f" - {issue_id} ({status})")
259 # Warn about cycles if detected
260 if has_cycles and dep_graph:
261 cycles = dep_graph.detect_cycles()
262 print("\nWarning: Dependency cycles detected:")
263 for cycle in cycles:
264 print(f" {' -> '.join(cycle)}")
266 # Render dependency analysis output
267 if dep_report is not None:
268 _render_dependency_analysis(
269 dep_report,
270 logger,
271 issue_to_wave=issue_to_wave if issue_to_wave else None,
272 config=dep_config,
273 )
275 if invalid:
276 print(f"\nWarning: {len(invalid)} issue(s) not found")
278 return 0
281# ---------------------------------------------------------------------------
282# Helper functions for enhanced show output (ENH-923)
283# ---------------------------------------------------------------------------
286def _format_created(iso_str: str) -> str:
287 """Format an ISO 8601 created timestamp as a human-friendly string."""
288 import time
289 from datetime import UTC, datetime
291 try:
292 dt = datetime.fromisoformat(iso_str)
293 if dt.tzinfo is None:
294 dt = dt.replace(tzinfo=UTC)
295 formatted = dt.strftime("%Y-%m-%d %H:%M UTC")
296 elapsed = time.time() - dt.timestamp()
297 if elapsed >= 0:
298 return f"{formatted} ({format_relative_time(elapsed)})"
299 return formatted
300 except (ValueError, OSError):
301 return iso_str
304def _print_composition(issue_infos: list[Any]) -> None:
305 """Print type/priority composition breakdown."""
306 if not issue_infos:
307 return
308 from collections import Counter
310 types: Counter[str] = Counter()
311 priorities: Counter[str] = Counter()
312 for info in issue_infos:
313 issue_type = info.issue_id.split("-", 1)[0]
314 types[issue_type] += 1
315 if info.priority:
316 priorities[info.priority] += 1
318 type_parts = [f"{count} {t}" for t, count in sorted(types.items())]
319 prio_parts = [f"{p}: {count}" for p, count in sorted(priorities.items())]
320 print(f"Composition: {', '.join(type_parts)} | {', '.join(prio_parts)}")
323def _print_run_state(sprint_name: str) -> None:
324 """Print last run state if .sprint-state.json exists for this sprint."""
325 import json
327 state_file = Path.cwd() / ".sprint-state.json"
328 if not state_file.exists():
329 return
330 try:
331 data = json.loads(state_file.read_text())
332 if data.get("sprint_name") != sprint_name:
333 return
334 completed = data.get("completed_issues", [])
335 failed = data.get("failed_issues", {})
336 skipped = data.get("skipped_blocked_issues", {})
337 total = len(completed) + len(failed) + len(skipped)
339 started = data.get("started_at", "")
340 date_str = started[:10] if started else "unknown"
342 parts = [f"{len(completed)} completed"]
343 if failed:
344 failed_ids = ", ".join(sorted(failed.keys()))
345 parts.append(f"{len(failed)} failed ({failed_ids})")
346 if skipped:
347 parts.append(f"{len(skipped)} skipped")
348 print(f"Last run: {date_str} \u2014 {', '.join(parts)} of {total}")
349 except (json.JSONDecodeError, OSError):
350 pass
353def _show_json(
354 sprint: Any,
355 issue_infos: list[Any],
356 waves: list[list[Any]],
357 contention_notes: Any,
358 has_cycles: bool,
359 dep_graph: Any,
360) -> int:
361 """Render sprint show output as JSON."""
362 issues_data = []
363 for info in issue_infos:
364 issues_data.append(
365 {
366 "id": info.issue_id,
367 "title": info.title,
368 "priority": info.priority,
369 "path": str(info.path),
370 "confidence_score": info.confidence_score,
371 "outcome_confidence": info.outcome_confidence,
372 }
373 )
375 waves_data = []
376 for wave_idx, wave in enumerate(waves):
377 waves_data.append(
378 {
379 "wave": wave_idx + 1,
380 "issues": [issue.issue_id for issue in wave],
381 }
382 )
384 data = {
385 "name": sprint.name,
386 "description": sprint.description or None,
387 "created": sprint.created,
388 "issues": issues_data,
389 "waves": waves_data,
390 "has_cycles": has_cycles,
391 }
392 print_json(data)
393 return 0