Coverage for little_loops / cli / issues / set_status.py: 0%

1"""ll-issues set-status: Transition an issue to a new status value.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import sys 

7from typing import TYPE_CHECKING 

8 

9if TYPE_CHECKING: 

10 from little_loops.config import BRConfig 

11 

12 

13def cmd_set_status(config: BRConfig, args: argparse.Namespace) -> int: 

14 """Write a new status value into an issue's YAML frontmatter. 

15 

16 Validates the target status against the canonical enum before writing. 

17 Prints the before→after transition to stdout on success. 

18 

19 When ``--cascade`` is set, also propagates the status to active children 

20 (those with status ``open``, ``in_progress``, or ``blocked``). Child 

21 resolution follows ``parent:`` edges **only**, transitively. Association 

22 edges (``relates_to:``, ``blocked_by:``) are non-hierarchical and never 

23 trigger a cascade — cascading through them silently mutated unrelated 

24 issues, including sibling epics (BUG-2265). 

25 

26 Args: 

27 config: Project configuration 

28 args: Parsed arguments with .issue_id, .status, .cascade, .cascade_to 

29 

30 Returns: 

31 Exit code (0 = success, 1 = error); exit 1 if any child update fails. 

32 """ 

33 from little_loops.cli.issues.show import _resolve_issue_id 

34 from little_loops.frontmatter import parse_frontmatter, update_frontmatter 

35 from little_loops.issue_progress import _OPEN_STATUSES, _TERMINAL_STATUSES 

36 

37 path = _resolve_issue_id(config, args.issue_id) 

38 if path is None: 

39 print(f"Error: Issue '{args.issue_id}' not found.", file=sys.stderr) 

40 return 1 

41 

42 # Validate cascade before making any changes 

43 if getattr(args, "cascade", False): 

44 if args.status not in _TERMINAL_STATUSES: 

45 print( 

46 f"Error: --cascade is only valid when target status is done or " 

47 f"cancelled, got '{args.status}'.", 

48 file=sys.stderr, 

49 ) 

50 return 1 

51 

52 content = path.read_text() 

53 old_status = parse_frontmatter(content).get("status", "unknown") 

54 new_content = update_frontmatter(content, {"status": args.status}) 

55 path.write_text(new_content) 

56 print(f"{args.issue_id}: {old_status} → {args.status}") 

57 

58 # Capture content snapshot on status transition (Decision 2: Option C — direct call, 

59 # same pattern as user_prompt_submit.py calling record_correction() without EventBus). 

60 try: 

61 from little_loops.session_store import record_issue_snapshot, resolve_history_db 

62 

63 db_path = resolve_history_db() 

64 record_issue_snapshot(db_path, args.issue_id, args.status, str(path)) 

65 except Exception: 

66 pass 

67 

68 # Cascade to children 

69 if getattr(args, "cascade", False): 

70 fm = parse_frontmatter(content) 

71 epic_id = fm.get("id", args.issue_id).upper() 

72 

73 from little_loops.issue_parser import find_issues 

74 

75 all_issues = find_issues(config) 

76 

77 # Cascade follows parent: → child edges ONLY, transitively. relates_to: 

78 # and blocked_by: are non-hierarchical association edges; cascading 

79 # through them silently flipped the status of unrelated issues — 

80 # including sibling epics — during routine epic closure (BUG-2265). 

81 children_by_parent: dict[str, list] = {} 

82 for i in all_issues: 

83 if i.parent: 

84 children_by_parent.setdefault(i.parent.upper(), []).append(i) 

85 

86 # Transitive closure over parent edges, breadth-first from the epic. 

87 descendants: list = [] 

88 seen: set[str] = {epic_id} 

89 queue = list(children_by_parent.get(epic_id, [])) 

90 while queue: 

91 child = queue.pop(0) 

92 cid = child.issue_id.upper() 

93 if cid in seen: 

94 continue 

95 seen.add(cid) 

96 descendants.append(child) 

97 queue.extend(children_by_parent.get(cid, [])) 

98 

99 active = [c for c in descendants if c.status in _OPEN_STATUSES] 

100 skipped = [c for c in descendants if c not in active] 

101 

102 print(f" Cascading to {len(active)} active parent-children (default: {args.cascade_to}):") 

103 

104 failures = 0 

105 for child in active: 

106 try: 

107 child_content = child.path.read_text() 

108 child_new = update_frontmatter(child_content, {"status": args.cascade_to}) 

109 child.path.write_text(child_new) 

110 print(f" {child.issue_id} → {args.cascade_to}") 

111 except OSError as exc: 

112 print(f" {child.issue_id}: FAILED ({exc})", file=sys.stderr) 

113 failures += 1 

114 

115 if skipped: 

116 print(f" ({len(skipped)} children already terminal/other — unchanged)") 

117 

118 if failures: 

119 return 1 

120 

121 return 0