Coverage for little_loops / hooks / sweep_stale_refs.py: 0%

1"""SessionEnd hook handler: sweep stale cross-issue status references — FEAT-1680. 

2 

3Fires at the end of every Claude Code session. Collects all issue IDs with 

4``status: done``, then scans open issue files for prose that still asserts 

5those IDs are ``open``, ``in_progress``, or active blockers. Findings are 

6reported via ``LLHookResult.feedback`` (stderr). When ``hooks.stale_ref_fix`` 

7is ``"auto"`` in ``ll-config.json``, stale phrases are rewritten in-place. 

8 

9The handler exits 0 in all cases — it is advisory only and must never block 

10session end. 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16import re 

17from pathlib import Path 

18 

19from little_loops.config.core import BRConfig, resolve_config_path 

20from little_loops.file_utils import atomic_write 

21from little_loops.hooks.types import LLHookEvent, LLHookResult 

22from little_loops.issue_parser import find_issues 

23from little_loops.text_utils import _CODE_FENCE 

24 

25# Matches any issue ID token (e.g. FEAT-1112, ENH-42, BUG-007, EPIC-3) 

26_ISSUE_ID_RE = re.compile(r"\b(ENH|BUG|FEAT|EPIC)-(\d+)\b") 

27 

28# Stale-status phrase patterns — matched *after* the issue ID on the same line 

29_STALE_STATUS_RE = re.compile( 

30 r"\bis\s+(?:still\s+)?(open|in_progress|active)\b", 

31 re.IGNORECASE, 

32) 

33 

34# Blocked-by phrase — the issue ID appears after "blocked by" 

35_BLOCKED_BY_RE = re.compile(r"\bblocked\s+by\b", re.IGNORECASE) 

36 

37 

38def _in_fence(pos_start: int, pos_end: int, fence_spans: list[tuple[int, int]]) -> bool: 

39 """Return True if the span [pos_start, pos_end) falls inside a code fence.""" 

40 return any(fs <= pos_start and pos_end <= fe for fs, fe in fence_spans) 

41 

42 

43def _scan_file( 

44 path: Path, 

45 done_ids: set[str], 

46) -> list[tuple[int, str, str]]: 

47 """Scan *path* for stale references to any ID in *done_ids*. 

48 

49 Returns a list of ``(lineno, issue_id, snippet)`` tuples (1-based lineno). 

50 Lines inside code fences are skipped. 

51 """ 

52 try: 

53 content = path.read_text(encoding="utf-8", errors="replace") 

54 except OSError: 

55 return [] 

56 

57 fence_spans = [(m.start(), m.end()) for m in _CODE_FENCE.finditer(content)] 

58 

59 findings: list[tuple[int, str, str]] = [] 

60 offset = 0 

61 for lineno, line in enumerate(content.splitlines(), start=1): 

62 line_start = offset 

63 line_end = offset + len(line) 

64 offset = line_end + 1 # +1 for the newline 

65 

66 if _in_fence(line_start, line_end, fence_spans): 

67 continue 

68 

69 for m in _ISSUE_ID_RE.finditer(line): 

70 issue_id = m.group(0) 

71 if issue_id not in done_ids: 

72 continue 

73 

74 # Check for stale-status phrase anywhere on the line 

75 if _STALE_STATUS_RE.search(line): 

76 snippet = line.strip() 

77 findings.append((lineno, issue_id, snippet)) 

78 break # one finding per line is enough 

79 

80 # Check for "blocked by <ID>" where this ID is done 

81 # Pattern: "blocked by" appears before the ID on the line 

82 pre_id = line[: m.start()] 

83 if _BLOCKED_BY_RE.search(pre_id): 

84 snippet = line.strip() 

85 findings.append((lineno, issue_id, snippet)) 

86 break 

87 

88 return findings 

89 

90 

91def _auto_fix_file(path: Path, done_ids: set[str]) -> bool: 

92 """Rewrite stale status phrases in *path* for any done ID mentioned. 

93 

94 Uses :func:`atomic_write` for safe in-place replacement. Returns True if 

95 the file was modified. 

96 """ 

97 try: 

98 content = path.read_text(encoding="utf-8", errors="replace") 

99 except OSError: 

100 return False 

101 

102 fence_spans = [(m.start(), m.end()) for m in _CODE_FENCE.finditer(content)] 

103 

104 new_lines: list[str] = [] 

105 modified = False 

106 offset = 0 

107 for line in content.splitlines(keepends=True): 

108 line_start = offset 

109 line_end = offset + len(line) 

110 offset = line_end 

111 

112 if _in_fence(line_start, line_end, fence_spans): 

113 new_lines.append(line) 

114 continue 

115 

116 # Only rewrite lines that mention a done ID 

117 has_done_ref = any(m.group(0) in done_ids for m in _ISSUE_ID_RE.finditer(line)) 

118 if has_done_ref and _STALE_STATUS_RE.search(line): 

119 new_line = _STALE_STATUS_RE.sub("is done", line) 

120 new_lines.append(new_line) 

121 modified = True 

122 else: 

123 new_lines.append(line) 

124 

125 if modified: 

126 atomic_write(path, "".join(new_lines)) 

127 return modified 

128 

129 

130def handle(event: LLHookEvent) -> LLHookResult: 

131 """Sweep open issue files for stale references to done issue IDs. 

132 

133 Reads ``hooks.stale_ref_fix`` from ``ll-config.json`` (default 

134 ``"report"``). When ``"auto"``, stale phrases are rewritten in-place. 

135 Always returns ``LLHookResult(exit_code=0)`` — findings are advisory and 

136 must never block session end. 

137 """ 

138 try: 

139 payload = event.payload or {} 

140 raw_cwd = payload.get("cwd") or (event.cwd or "") 

141 cwd = Path(raw_cwd) if raw_cwd else Path.cwd() 

142 

143 # Load raw hooks config to read stale_ref_fix mode 

144 config_path = resolve_config_path(cwd) 

145 fix_mode = "report" 

146 if config_path is not None: 

147 try: 

148 raw_config = json.loads(config_path.read_text(encoding="utf-8")) 

149 raw_hooks = raw_config.get("hooks", {}) 

150 if isinstance(raw_hooks, dict): 

151 fix_mode = raw_hooks.get("stale_ref_fix", "report") 

152 except (OSError, json.JSONDecodeError): 

153 pass 

154 

155 # Build BRConfig for find_issues() 

156 config = BRConfig(project_root=cwd) 

157 

158 # Collect done IDs 

159 done_issues = find_issues(config, status_filter={"done"}) 

160 done_ids: set[str] = {i.issue_id for i in done_issues} 

161 

162 if not done_ids: 

163 return LLHookResult(exit_code=0) 

164 

165 # Collect open issues (default call skips done/cancelled/deferred) 

166 open_issues = find_issues(config) 

167 

168 all_findings: list[tuple[Path, int, str, str]] = [] 

169 

170 for issue in open_issues: 

171 path: Path = issue.path 

172 if not path.is_file(): 

173 continue 

174 

175 if fix_mode == "auto": 

176 _auto_fix_file(path, done_ids) 

177 

178 findings = _scan_file(path, done_ids) 

179 for lineno, issue_id, snippet in findings: 

180 all_findings.append((path, lineno, issue_id, snippet)) 

181 

182 if not all_findings: 

183 return LLHookResult(exit_code=0) 

184 

185 lines = [f"[ll] {len(all_findings)} stale cross-issue reference(s) found:"] 

186 for path, lineno, issue_id, snippet in all_findings: 

187 lines.append(f" {path}:{lineno}: [{issue_id}] {snippet}") 

188 feedback = "\n".join(lines) 

189 

190 return LLHookResult(exit_code=0, feedback=feedback) 

191 

192 except Exception: 

193 return LLHookResult(exit_code=0)