Coverage for little_loops / cli / deps.py: 2%
220 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:19 -0500
1"""ll-deps: Cross-issue dependency discovery and validation."""
3from __future__ import annotations
5import argparse
6from pathlib import Path
9def _load_issues(
10 issues_dir: Path,
11 only_ids: set[str] | None = None,
12) -> tuple[list, dict[str, str], set[str]]:
13 """Load issues from directory for CLI use.
15 Args:
16 issues_dir: Path to the issues base directory (e.g., .issues)
17 only_ids: If provided, only include issues with these IDs
19 Returns:
20 Tuple of (active issues, issue contents map, completed issue IDs)
21 """
22 from little_loops.config import BRConfig
23 from little_loops.issue_parser import find_issues
25 # Find project root by walking up from issues_dir
26 project_root = issues_dir.resolve().parent
27 if issues_dir.name != ".issues":
28 # If issues_dir is already absolute, try to find config relative to it
29 project_root = issues_dir.parent
31 config = BRConfig(project_root)
32 issues = find_issues(config, only_ids=only_ids)
34 # Build contents map
35 issue_contents: dict[str, str] = {}
36 for info in issues:
37 if info.path.exists():
38 issue_contents[info.issue_id] = info.path.read_text(encoding="utf-8")
40 # Gather completed and deferred issue IDs
41 import re as _re
43 completed_ids: set[str] = set()
44 for non_active_dir in [config.get_completed_dir(), config.get_deferred_dir()]:
45 if non_active_dir.exists():
46 for f in non_active_dir.glob("*.md"):
47 match = _re.search(r"(BUG|FEAT|ENH)-(\d+)", f.name)
48 if match:
49 completed_ids.add(f"{match.group(1)}-{match.group(2)}")
51 return issues, issue_contents, completed_ids
54def main_deps() -> int:
55 """Entry point for ll-deps command.
57 Analyze cross-issue dependencies and validate existing references.
59 Returns:
60 Exit code (0 = success, 1 = failure)
61 """
62 import json as _json
63 import sys
65 from little_loops.dependency_mapper import (
66 analyze_dependencies,
67 fix_dependencies,
68 format_report,
69 format_text_graph,
70 gather_all_issue_ids,
71 validate_dependencies,
72 )
74 parser = argparse.ArgumentParser(
75 prog="ll-deps",
76 description="Cross-issue dependency discovery and validation",
77 formatter_class=argparse.RawDescriptionHelpFormatter,
78 epilog="""
79Examples:
80 %(prog)s analyze # Full analysis with markdown output
81 %(prog)s analyze --format json # JSON output for programmatic use
82 %(prog)s analyze --graph # Include ASCII dependency graph
83 %(prog)s analyze --sprint my-sprint # Analyze only issues in a sprint
84 %(prog)s validate # Validation only (broken refs, cycles)
85 %(prog)s validate --sprint my-sprint # Validate only sprint issue deps
86 %(prog)s fix # Auto-fix broken refs, stale refs, backlinks
87 %(prog)s fix --dry-run # Preview fixes without modifying files
88 %(prog)s apply # Apply proposals >= 0.7 confidence
89 %(prog)s apply --min-confidence 0.5 # Lower threshold
90 %(prog)s apply --dry-run # Preview only (no writes)
91 %(prog)s apply --sprint my-sprint # Sprint-scoped apply
92 %(prog)s apply FEAT-001 blocks FEAT-002 # Manual explicit pair
93 %(prog)s apply FEAT-001 blocked-by FEAT-002 # Manual explicit pair (inverse)
94""",
95 )
97 parser.add_argument(
98 "-d",
99 "--issues-dir",
100 type=Path,
101 default=None,
102 help="Path to issues directory (default: .issues)",
103 )
105 subparsers = parser.add_subparsers(dest="command", help="Available commands")
107 # analyze subcommand
108 analyze_parser = subparsers.add_parser(
109 "analyze",
110 help="Full dependency analysis (file overlaps + validation)",
111 )
112 analyze_parser.add_argument(
113 "-f",
114 "--format",
115 type=str,
116 choices=["text", "json"],
117 default="text",
118 help="Output format (default: text/markdown)",
119 )
120 analyze_parser.add_argument(
121 "--graph",
122 action="store_true",
123 help="Include ASCII dependency graph in output",
124 )
125 analyze_parser.add_argument(
126 "--sprint",
127 type=str,
128 default=None,
129 help="Restrict analysis to issues in the named sprint",
130 )
132 # validate subcommand
133 validate_parser = subparsers.add_parser(
134 "validate",
135 help="Validate existing dependency references only",
136 )
137 validate_parser.add_argument(
138 "--sprint",
139 type=str,
140 default=None,
141 help="Restrict validation to issues in the named sprint",
142 )
144 # fix subcommand
145 fix_parser = subparsers.add_parser(
146 "fix",
147 help="Auto-fix broken refs, stale refs, and missing backlinks",
148 )
149 fix_parser.add_argument(
150 "--dry-run",
151 "-n",
152 action="store_true",
153 help="Show what would be fixed without making changes",
154 )
155 fix_parser.add_argument(
156 "--sprint",
157 type=str,
158 default=None,
159 help="Restrict fixes to issues in the named sprint",
160 )
162 # apply subcommand
163 apply_parser = subparsers.add_parser(
164 "apply",
165 help="Write dependency relationships to issue files",
166 )
167 apply_parser.add_argument(
168 "source",
169 nargs="?",
170 default=None,
171 help="Source issue ID for explicit pair (e.g. FEAT-001)",
172 )
173 apply_parser.add_argument(
174 "relation",
175 nargs="?",
176 default=None,
177 choices=["blocks", "blocked-by"],
178 help="Relationship direction: 'blocks' or 'blocked-by'",
179 )
180 apply_parser.add_argument(
181 "target",
182 nargs="?",
183 default=None,
184 help="Target issue ID for explicit pair (e.g. FEAT-002)",
185 )
186 apply_parser.add_argument(
187 "--min-confidence",
188 type=float,
189 default=0.7,
190 help="Minimum confidence threshold for implicit apply (default: 0.7)",
191 )
192 apply_parser.add_argument(
193 "--dry-run",
194 "-n",
195 action="store_true",
196 help="Preview without writing",
197 )
198 apply_parser.add_argument(
199 "--sprint",
200 type=str,
201 default=None,
202 help="Restrict to issues in named sprint",
203 )
205 args = parser.parse_args()
207 if not args.command:
208 parser.print_help()
209 return 1
211 issues_dir = args.issues_dir or Path.cwd() / ".issues"
212 if not issues_dir.exists():
213 print(f"Error: Issues directory not found: {issues_dir}", file=sys.stderr)
214 return 1
216 # Sprint-scoped filtering
217 only_ids: set[str] | None = None
218 if getattr(args, "sprint", None):
219 from little_loops.config import BRConfig as _BRConfig
220 from little_loops.sprint import Sprint
222 project_root = issues_dir.resolve().parent
223 if issues_dir.name != ".issues":
224 project_root = issues_dir.parent
225 _config = _BRConfig(project_root)
226 sprints_dir = Path(_config.sprints.sprints_dir)
227 if not sprints_dir.is_absolute():
228 sprints_dir = project_root / sprints_dir
230 sprint = Sprint.load(sprints_dir, args.sprint)
231 if sprint is None:
232 print(f"Error: Sprint not found: {args.sprint}", file=sys.stderr)
233 return 1
234 only_ids = set(sprint.issues)
235 if not only_ids:
236 print(f"Sprint '{args.sprint}' has no issues.")
237 return 0
239 try:
240 issues, issue_contents, completed_ids = _load_issues(issues_dir, only_ids=only_ids)
241 except Exception as e:
242 print(f"Error loading issues: {e}", file=sys.stderr)
243 return 1
245 if not issues:
246 print("No active issues found.")
247 return 0
249 # Gather all issue IDs on disk to avoid false "nonexistent" warnings
250 # when sprint-scoped analysis references issues outside the sprint
251 try:
252 from little_loops.config import BRConfig as _BRConfig
254 _dm_config = _BRConfig(issues_dir.resolve().parent)
255 except Exception:
256 _dm_config = None
257 all_known_ids = gather_all_issue_ids(issues_dir, config=_dm_config)
259 # Load dependency mapping config
260 dep_config = _dm_config.dependency_mapping if _dm_config else None
262 if args.command == "analyze":
263 report = analyze_dependencies(
264 issues, issue_contents, completed_ids, all_known_ids, config=dep_config
265 )
267 if args.format == "json":
268 data = {
269 "issue_count": report.issue_count,
270 "existing_dep_count": report.existing_dep_count,
271 "proposals": [
272 {
273 "source_id": p.source_id,
274 "target_id": p.target_id,
275 "reason": p.reason,
276 "confidence": p.confidence,
277 "rationale": p.rationale,
278 "overlapping_files": p.overlapping_files,
279 "conflict_score": p.conflict_score,
280 }
281 for p in report.proposals
282 ],
283 "parallel_safe": [
284 {
285 "issue_a": ps.issue_a,
286 "issue_b": ps.issue_b,
287 "shared_files": ps.shared_files,
288 "conflict_score": ps.conflict_score,
289 "reason": ps.reason,
290 }
291 for ps in report.parallel_safe
292 ],
293 "validation": {
294 "broken_refs": report.validation.broken_refs,
295 "missing_backlinks": report.validation.missing_backlinks,
296 "cycles": report.validation.cycles,
297 "stale_completed_refs": report.validation.stale_completed_refs,
298 "has_issues": report.validation.has_issues,
299 },
300 }
301 print(_json.dumps(data, indent=2))
302 else:
303 print(format_report(report, config=dep_config))
304 if args.graph:
305 print()
306 print("## Dependency Graph")
307 print()
308 print(format_text_graph(issues, report.proposals))
310 return 0
312 if args.command == "validate":
313 result = validate_dependencies(issues, completed_ids, all_known_ids)
315 if not result.has_issues:
316 print("No validation issues found.")
317 return 0
319 lines: list[str] = []
320 lines.append("# Dependency Validation Report")
321 lines.append("")
323 if result.broken_refs:
324 lines.append("## Broken References")
325 lines.append("")
326 for issue_id, ref_id in result.broken_refs:
327 lines.append(f"- {issue_id}: references nonexistent {ref_id}")
328 lines.append("")
330 if result.missing_backlinks:
331 lines.append("## Missing Backlinks")
332 lines.append("")
333 for issue_id, ref_id in result.missing_backlinks:
334 lines.append(
335 f"- {issue_id} is blocked by {ref_id}, "
336 f"but {ref_id} does not list {issue_id} in Blocks"
337 )
338 lines.append("")
340 if result.cycles:
341 lines.append("## Dependency Cycles")
342 lines.append("")
343 for cycle in result.cycles:
344 lines.append(f"- {' -> '.join(cycle)}")
345 lines.append("")
347 if result.stale_completed_refs:
348 lines.append("## Stale References (to completed issues)")
349 lines.append("")
350 for issue_id, ref_id in result.stale_completed_refs:
351 lines.append(f"- {issue_id}: blocked by {ref_id} (completed)")
352 lines.append("")
354 print("\n".join(lines))
355 return 0
357 if args.command == "fix":
358 fix_result = fix_dependencies(issues, completed_ids, all_known_ids, dry_run=args.dry_run)
360 if not fix_result.changes:
361 print("No fixable issues found.")
362 if fix_result.skipped_cycles:
363 print(f"({fix_result.skipped_cycles} cycle(s) detected — resolve manually)")
364 return 0
366 prefix = "[DRY RUN] " if args.dry_run else ""
367 print(f"# {prefix}Dependency Fix Report")
368 print()
369 for change in fix_result.changes:
370 print(f" {prefix}{change}")
371 print()
372 print(f"{prefix}{len(fix_result.changes)} fix(es) applied.")
374 if fix_result.modified_files:
375 print()
376 print("Modified files:")
377 for fpath in sorted(fix_result.modified_files):
378 print(f" {fpath}")
380 if fix_result.skipped_cycles:
381 print()
382 print(f"({fix_result.skipped_cycles} cycle(s) detected — resolve manually)")
384 return 0
386 if args.command == "apply":
387 from little_loops.dependency_mapper.operations import _add_to_section
389 prefix = "[DRY RUN] " if args.dry_run else ""
390 issue_files = {i.issue_id: i.path for i in issues}
392 # Explicit-pair mode: all three positional args must be provided together
393 if args.source or args.relation or args.target:
394 if not (args.source and args.relation and args.target):
395 print(
396 "Error: explicit pair requires all three arguments: "
397 "<source> <relation> <target>",
398 file=sys.stderr,
399 )
400 return 1
402 all_ids = {i.issue_id for i in issues} | all_known_ids
403 for id_to_check, label in [(args.source, "source"), (args.target, "target")]:
404 if id_to_check not in all_ids:
405 print(
406 f"Error: {label} issue {id_to_check!r} not found",
407 file=sys.stderr,
408 )
409 return 1
411 # Determine which issue receives the "Blocked By" entry
412 if args.relation == "blocks":
413 # source blocks target → target is blocked by source
414 blocked_id, blocker_id = args.target, args.source
415 else: # blocked-by
416 # source blocked-by target → source is blocked by target
417 blocked_id, blocker_id = args.source, args.target
419 blocked_path = issue_files.get(blocked_id)
420 if blocked_path is None:
421 print(
422 f"Error: issue {blocked_id!r} is not in active issues (cannot write to it)",
423 file=sys.stderr,
424 )
425 return 1
427 print(f"# {prefix}Dependency Apply Report")
428 print()
429 print(f" {prefix}{blocked_id} blocked by {blocker_id}")
430 print()
432 if not args.dry_run:
433 _add_to_section(blocked_path, "Blocked By", blocker_id)
434 print("1 relationship(s) applied.")
435 print()
436 print("Modified files:")
437 print(f" {blocked_path}")
438 else:
439 print("[DRY RUN] 1 relationship(s) would be applied.")
441 return 0
443 # Implicit mode: run analysis and apply proposals above confidence threshold
444 report = analyze_dependencies(
445 issues, issue_contents, completed_ids, all_known_ids, config=dep_config
446 )
447 filtered = [p for p in report.proposals if p.confidence >= args.min_confidence]
449 if not filtered:
450 print(f"No proposals at or above confidence threshold ({args.min_confidence}).")
451 return 0
453 print(f"# {prefix}Dependency Apply Report")
454 print()
456 modified: set[str] = set()
457 applied = 0
459 for proposal in filtered:
460 source_path = issue_files.get(proposal.source_id)
461 if source_path is None or not source_path.exists():
462 continue
463 desc = (
464 f"{proposal.source_id} blocked by {proposal.target_id}"
465 f" (confidence: {proposal.confidence:.2f})"
466 )
467 print(f" {prefix}{desc}")
468 applied += 1
470 if not args.dry_run:
471 _add_to_section(source_path, "Blocked By", proposal.target_id)
472 modified.add(str(source_path))
474 print()
475 if args.dry_run:
476 print(f"[DRY RUN] {applied} relationship(s) would be applied.")
477 else:
478 print(f"{applied} relationship(s) applied.")
479 if modified:
480 print()
481 print("Modified files:")
482 for fpath in sorted(modified):
483 print(f" {fpath}")
485 return 0
487 return 1