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

38 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2026-05-28 13:07 -0500

1"""PostToolUse hook handler: per-tool byte metrics for context-window analytics. 

2 

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

4(FEAT-1112) on every tool call. Each row captures ``bytes_in``, ``bytes_out``, 

5and ``cache_hit`` derived from the ``LLHookEvent`` payload so that downstream 

6consumers (FEAT-1624 ``/ll:ctx-stats``) can surface which tools consumed the 

7most context-window bytes during a session. 

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 

20from pathlib import Path 

21from typing import Any 

22 

23from little_loops.config.core import resolve_config_path 

24from little_loops.config.features import feature_enabled 

25from little_loops.hooks.types import LLHookEvent, LLHookResult 

26 

27 

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

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

30 config_path = resolve_config_path(cwd) 

31 if config_path is None: 

32 return None 

33 try: 

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

35 except (OSError, json.JSONDecodeError): 

36 return None 

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

38 

39 

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

41 """Persist per-tool byte metrics to ``tool_events`` (FEAT-1623). 

42 

43 Gated on ``analytics.enabled``; silent on any failure. 

44 """ 

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

46 config = _load_config(cwd) 

47 if config is None or not feature_enabled(config, "analytics.enabled"): 

48 return LLHookResult(exit_code=0) 

49 

50 payload = event.payload or {} 

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

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

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

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

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

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

57 session_id = payload.get("session_id") 

58 

59 with contextlib.suppress(Exception): 

60 from little_loops.session_store import _hash_args, _now, connect 

61 

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

63 try: 

64 conn.execute( 

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

66 "result_size, bytes_in, bytes_out, cache_hit) " 

67 "VALUES(?, ?, ?, ?, ?, ?, ?, ?)", 

68 ( 

69 _now(), 

70 session_id, 

71 tool_name, 

72 _hash_args(tool_input), 

73 bytes_out, 

74 bytes_in, 

75 bytes_out, 

76 cache_hit, 

77 ), 

78 ) 

79 conn.commit() 

80 finally: 

81 conn.close() 

82 

83 return LLHookResult(exit_code=0)