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 uses union of ``relates_to:`` (forward) + ``parent:`` (backward) 

22 lookups. 

23 

24 Args: 

25 config: Project configuration 

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

27 

28 Returns: 

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

30 """ 

31 from little_loops.cli.issues.show import _resolve_issue_id 

32 from little_loops.frontmatter import parse_frontmatter, update_frontmatter 

33 from little_loops.issue_progress import _OPEN_STATUSES, _TERMINAL_STATUSES 

34 

35 path = _resolve_issue_id(config, args.issue_id) 

36 if path is None: 

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

38 return 1 

39 

40 # Validate cascade before making any changes 

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

42 if args.status not in _TERMINAL_STATUSES: 

43 print( 

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

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

46 file=sys.stderr, 

47 ) 

48 return 1 

49 

50 content = path.read_text() 

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

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

53 path.write_text(new_content) 

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

55 

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

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

58 try: 

59 from little_loops.session_store import record_issue_snapshot, resolve_history_db 

60 

61 db_path = resolve_history_db() 

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

63 except Exception: 

64 pass 

65 

66 # Cascade to children 

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

68 fm = parse_frontmatter(content) 

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

70 

71 from little_loops.issue_parser import find_issues 

72 

73 all_issues = find_issues(config) 

74 

75 # Union of forward (relates_to:) + backward (parent:) 

76 forward_ids: set[str] = set(fm.get("relates_to", [])) 

77 backward_ids: set[str] = { 

78 i.issue_id for i in all_issues if i.parent and i.parent.upper() == epic_id 

79 } 

80 child_ids = forward_ids | backward_ids 

81 

82 children = [i for i in all_issues if i.issue_id.upper() in child_ids] 

83 active = [c for c in children if c.status in _OPEN_STATUSES] 

84 skipped = [c for c in children if c not in active] 

85 

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

87 

88 failures = 0 

89 for child in active: 

90 try: 

91 child_content = child.path.read_text() 

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

93 child.path.write_text(child_new) 

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

95 except OSError as exc: 

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

97 failures += 1 

98 

99 if skipped: 

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

101 

102 if failures: 

103 return 1 

104 

105 return 0