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

1"""PostToolUse hook handler: per-tool byte metrics and file-event recording. 

2 

3Persists a row into the ``tool_events`` table of the unified session store 

4(FEAT-1112) on every tool call, and a row into ``file_events`` for tool calls 

5that touch a file (ENH-1832). ``tool_events`` rows capture ``bytes_in``, 

6``bytes_out``, and ``cache_hit``; ``file_events`` rows capture the file path, 

7op (tool name), and optional issue ID extracted from the path. 

8 

9Guarded by the ``analytics.enabled`` config flag — when absent or false, the 

10handler is a no-op so projects that do not opt in pay no SQLite cost on the 

11hot tool-call path. SQLite failures (locked store, missing path, schema drift) 

12degrade silently: the ``__init__.main_hooks`` dispatcher has no try/except, so 

13any exception here would surface to the host as a hook failure. 

14""" 

15 

16from __future__ import annotations 

17 

18import contextlib 

19import json 

20import re 

21import subprocess 

22from pathlib import Path 

23from typing import Any 

24 

25from little_loops.config.core import resolve_config_path 

26from little_loops.config.features import AnalyticsCaptureConfig, feature_enabled 

27from little_loops.hooks.types import LLHookEvent, LLHookResult 

28 

29# Matches a file-path token inside a Bash command string. Two alternatives: 

30# 1. Paths that start with ./, ../, or / (absolute/explicit-relative) 

31# 2. Paths that start with a letter and contain at least one / (e.g. scripts/foo.py) 

32# Anchored by start-of-string or preceding whitespace so flag-like tokens 

33# (e.g. -v, --flag) are not mistaken for paths. 

34_BASH_PATH_RE = re.compile(r"(?:^|\s)((?:\./|\.\.\/|/)[^\s\"'|&;><]+|[A-Za-z][\w./\-]*/[\w./\-]+)") 

35 

36# Matches an issue ID (e.g. ENH-1832, BUG-007) within a file path. 

37_ISSUE_ID_RE = re.compile(r"((?:BUG|FEAT|ENH|EPIC)-\d+)") 

38 

39 

40def _load_config(cwd: Path) -> dict[str, Any] | None: 

41 """Load ``.ll/ll-config.json`` (host-aware), returning None on miss/error.""" 

42 config_path = resolve_config_path(cwd) 

43 if config_path is None: 

44 return None 

45 try: 

46 data = json.loads(config_path.read_text(encoding="utf-8")) 

47 except (OSError, json.JSONDecodeError): 

48 return None 

49 return data if isinstance(data, dict) else None 

50 

51 

52def _extract_file_path(tool_name: str, tool_input: dict[str, Any]) -> str | None: 

53 """Return the primary file path from a tool_input dict, or None if not applicable.""" 

54 if tool_name in {"Read", "Write", "Edit"}: 

55 return tool_input.get("file_path") or None 

56 if tool_name == "Glob": 

57 return tool_input.get("path") or tool_input.get("pattern") or None 

58 if tool_name == "Grep": 

59 return tool_input.get("path") or None 

60 if tool_name == "Bash": 

61 cmd = tool_input.get("command") or "" 

62 m = _BASH_PATH_RE.search(cmd) 

63 return m.group(1) if m else None 

64 return None 

65 

66 

67def _normalize_path(raw: str, cwd: Path) -> str: 

68 """Return path relative to cwd with no leading './'. 

69 

70 Relative paths are returned as-is (preserving trailing slash / glob patterns) 

71 after stripping a leading './'. Absolute paths are made relative to cwd when 

72 possible, falling back to the original string. 

73 """ 

74 p = Path(raw) 

75 if p.is_absolute(): 

76 try: 

77 return str(p.relative_to(cwd)) 

78 except ValueError: 

79 return raw 

80 return raw[2:] if raw.startswith("./") else raw 

81 

82 

83def _detect_issue_id(path: str) -> str | None: 

84 """Return an issue ID if the path points into .issues/, else None.""" 

85 if ".issues/" not in path: 

86 return None 

87 m = _ISSUE_ID_RE.search(path) 

88 return m.group(1) if m else None 

89 

90 

91def _maybe_auto_commit(config: dict[str, Any], cwd: Path, file_path: str, tool_name: str) -> None: 

92 """Auto-commit an issue file change when issues.auto_commit is enabled (ENH-1844).""" 

93 if not feature_enabled(config, "issues.auto_commit"): 

94 return 

95 

96 filename = Path(file_path).name 

97 if not re.match(r"^P[0-5]-(BUG|FEAT|ENH|EPIC)-\d{3,}", filename): 

98 return 

99 

100 issues_cfg = config.get("issues", {}) 

101 base_dir: str = issues_cfg.get("base_dir", ".issues") 

102 norm = file_path.replace("\\", "/") 

103 if f"/{base_dir}/" not in norm and not norm.startswith(f"{base_dir}/"): 

104 return 

105 

106 prefix: str = issues_cfg.get("auto_commit_prefix", "chore(issues)") 

107 m = re.match(r"^P[0-5]-((BUG|FEAT|ENH|EPIC)-\d+)-(.+?)\.md$", filename) 

108 if not m: 

109 return 

110 issue_id = m.group(1) 

111 slug = m.group(3) 

112 verb = "capture" if tool_name == "Write" else "update" 

113 commit_msg = f"{prefix}: {verb} {issue_id} {slug}" 

114 

115 abs_path = Path(file_path) if Path(file_path).is_absolute() else (cwd / file_path) 

116 

117 with contextlib.suppress(Exception): 

118 subprocess.run(["git", "add", str(abs_path)], cwd=str(cwd), check=True, capture_output=True) 

119 status_result = subprocess.run( 

120 ["git", "status", "--porcelain"], 

121 cwd=str(cwd), 

122 capture_output=True, 

123 text=True, 

124 check=True, 

125 ) 

126 other_changes = [ln for ln in status_result.stdout.splitlines() if filename not in ln] 

127 if other_changes: 

128 return 

129 subprocess.run( 

130 ["git", "commit", "-m", commit_msg], 

131 cwd=str(cwd), 

132 check=True, 

133 capture_output=True, 

134 ) 

135 

136 

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

138 """Persist per-tool byte metrics and auto-commit issue files when configured. 

139 

140 Analytics writes are gated on ``analytics.enabled``; auto-commit is gated on 

141 ``issues.auto_commit``. Both features degrade silently on failure. 

142 """ 

143 cwd = Path(event.cwd) if event.cwd else Path.cwd() 

144 config = _load_config(cwd) 

145 

146 payload = event.payload or {} 

147 tool_input = payload.get("tool_input", {}) or {} 

148 tool_name = str(payload.get("tool_name", "")) 

149 raw_path = _extract_file_path(tool_name, tool_input) 

150 

151 if config is not None and feature_enabled(config, "analytics.enabled"): 

152 tool_response = payload.get("tool_response", {}) or {} 

153 bytes_in = len(json.dumps(tool_input, default=str)) 

154 bytes_out = len(json.dumps(tool_response, default=str)) 

155 cache_hit = 1 if payload.get("cache_hit") else 0 

156 session_id = payload.get("session_id") 

157 

158 with contextlib.suppress(Exception): 

159 from little_loops.session_store import _hash_args, _now, connect 

160 

161 conn = connect(cwd / ".ll" / "history.db") 

162 try: 

163 conn.execute( 

164 "INSERT INTO tool_events(ts, session_id, tool_name, args_hash, " 

165 "result_size, bytes_in, bytes_out, cache_hit) " 

166 "VALUES(?, ?, ?, ?, ?, ?, ?, ?)", 

167 ( 

168 _now(), 

169 session_id, 

170 tool_name, 

171 _hash_args(tool_input), 

172 bytes_out, 

173 bytes_in, 

174 bytes_out, 

175 cache_hit, 

176 ), 

177 ) 

178 conn.commit() 

179 finally: 

180 conn.close() 

181 

182 if raw_path: 

183 capture = AnalyticsCaptureConfig.from_dict( 

184 config.get("analytics", {}).get("capture", {}) 

185 ) 

186 if capture.file_events: 

187 with contextlib.suppress(Exception): 

188 from little_loops.session_store import write_file_event 

189 

190 norm_path = _normalize_path(raw_path, cwd) 

191 issue_id = _detect_issue_id(norm_path) 

192 write_file_event( 

193 cwd / ".ll" / "history.db", 

194 session_id, 

195 norm_path, 

196 tool_name, 

197 issue_id, 

198 ) 

199 # TODO(ENH-1835): wire analytics.capture.skills gate when ENH-1833 lands 

200 

201 if config is not None and tool_name in {"Write", "Edit"} and raw_path: 

202 _maybe_auto_commit(config, cwd, raw_path, tool_name) 

203 

204 if tool_name == "Bash": 

205 from little_loops.hooks import install_learning_gate 

206 

207 return install_learning_gate.gate(event) 

208 

209 return LLHookResult(exit_code=0)