Coverage for little_loops / cli / ctx_stats.py: 13%
214 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:30 -0500
1"""ll-ctx-stats: Context-window analytics for the current project (FEAT-1624).
3Reads per-tool byte metrics that the ``post_tool_use`` hook persists into
4``.ll/history.db`` (FEAT-1623) and renders a compact summary of how much
5data was processed by tools vs. how much actually entered the conversation
6context. Falls back to ``.ll/ll-context-state.json`` (token estimates) when
7the SQLite store is absent so first-time users still get useful output.
8"""
10from __future__ import annotations
12import argparse
13import json
14import sqlite3
15import sys
16from collections import defaultdict
17from pathlib import Path
18from typing import Any
20from little_loops.cli.logs import _aggregate_skill_stats
21from little_loops.cli.output import (
22 configure_output,
23 format_relative_time,
24 terminal_width,
25 use_color_enabled,
26)
27from little_loops.logger import Logger
28from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
29from little_loops.user_messages import get_project_folder
31DEFAULT_DB_RELPATH = Path(".ll") / "history.db"
32DEFAULT_STATE_RELPATH = Path(".ll") / "ll-context-state.json"
35def _build_parser() -> argparse.ArgumentParser:
36 """Build the ll-ctx-stats argument parser (exposed for testing)."""
37 parser = argparse.ArgumentParser(
38 prog="ll-ctx-stats",
39 description=(
40 "Show context-window savings metrics and skill-health signals for the current project. "
41 "Reads per-tool byte metrics from .ll/history.db and renders how much data was processed "
42 "by tools vs. how much entered conversation context. Also surfaces per-skill invocation "
43 "frequency and correction rate from the same database."
44 ),
45 formatter_class=argparse.RawDescriptionHelpFormatter,
46 epilog="""
47Examples:
48 %(prog)s # Print savings summary and skill-health section
49 %(prog)s --db PATH # Use a non-default session database
50 %(prog)s --json # Output as JSON (includes skill_health array)
52Exit codes:
53 0 - Report rendered (data present or fallback used)
54 1 - No data found in either the SQLite store or the fallback file
55""",
56 )
57 parser.add_argument(
58 "--db",
59 type=Path,
60 default=None,
61 help="Path to the session database (default: .ll/history.db)",
62 )
63 parser.add_argument(
64 "-j",
65 "--json",
66 dest="json_mode",
67 action="store_true",
68 help="Output as JSON",
69 )
70 return parser
73def _parse_args(argv: list[str] | None) -> argparse.Namespace:
74 """Parse argv into a Namespace (exposed for testing)."""
75 return _build_parser().parse_args(argv)
78def _format_bytes(value: int) -> str:
79 """Render *value* bytes as a short ``KB``/``MB`` string."""
80 if value < 1024:
81 return f"{value} B"
82 if value < 1024 * 1024:
83 return f"{value / 1024:.1f} KB"
84 return f"{value / (1024 * 1024):.1f} MB"
87def _time_gained(seconds: float) -> str:
88 """Render *seconds* as a positive-tense ``+Xm`` string.
90 ``format_relative_time`` appends ``" ago"`` (it is designed for past-tense
91 durations); strip that suffix here so the line reads as savings rather
92 than elapsed time. The shared helper is intentionally left unchanged
93 (see Implementation Constraints #4 on FEAT-1624).
94 """
95 label = format_relative_time(seconds)
96 if label.endswith(" ago"):
97 label = label[: -len(" ago")]
98 return f"+{label}"
101def _progress_bar(value: int, ceiling: int, width: int) -> str:
102 """Return a ``|#### |`` bar of ``width`` columns scaled to ``value/ceiling``."""
103 if width < 3:
104 width = 3
105 inner = width - 2
106 if ceiling <= 0:
107 filled = 0
108 else:
109 filled = max(0, min(inner, round(inner * value / ceiling)))
110 return "|" + "#" * filled + " " * (inner - filled) + "|"
113def _aggregate_tool_events(db_path: Path) -> dict[str, Any] | None:
114 """Sum per-tool byte metrics from ``tool_events``.
116 Backfilled rows have ``NULL`` byte columns (see Implementation Constraints
117 #1 on FEAT-1624). Per-tool aggregation filters those rows out so historic
118 JSONL noise does not skew the summary; cache totals likewise.
120 Returns ``None`` when the database file is missing. Returns an empty
121 summary (all zeros) when the database exists but has no analytic rows.
122 """
123 if not db_path.exists():
124 return None
125 conn = sqlite3.connect(str(db_path))
126 try:
127 conn.row_factory = sqlite3.Row
128 try:
129 rows = conn.execute(
130 "SELECT tool_name, bytes_in, bytes_out, cache_hit "
131 "FROM tool_events WHERE bytes_in IS NOT NULL OR bytes_out IS NOT NULL"
132 ).fetchall()
133 except sqlite3.OperationalError:
134 return None
135 finally:
136 conn.close()
138 per_tool: dict[str, dict[str, int]] = defaultdict(lambda: {"calls": 0, "bytes": 0})
139 total_in = 0
140 total_out = 0
141 cache_hits = 0
142 cache_bytes = 0
143 for row in rows:
144 tool = (row["tool_name"] or "unknown").lower()
145 bin_ = int(row["bytes_in"] or 0)
146 bout = int(row["bytes_out"] or 0)
147 per_tool[tool]["calls"] += 1
148 per_tool[tool]["bytes"] += bout
149 total_in += bin_
150 total_out += bout
151 if row["cache_hit"]:
152 cache_hits += 1
153 cache_bytes += bout
155 return {
156 "total_in": total_in,
157 "total_out": total_out,
158 "cache_hits": cache_hits,
159 "cache_bytes": cache_bytes,
160 "per_tool": dict(per_tool),
161 }
164def _load_fallback_state(path: Path) -> dict[str, Any] | None:
165 """Return ``.ll/ll-context-state.json`` parsed, or ``None`` if absent/invalid."""
166 if not path.exists():
167 return None
168 try:
169 data = json.loads(path.read_text(encoding="utf-8"))
170 except (OSError, json.JSONDecodeError):
171 return None
172 return data if isinstance(data, dict) else None
175def _compute_cache_rate_from_jsonl(cwd: Path) -> dict[str, Any] | None:
176 """Compute session-aggregate cache hit rate from the most recent JSONL transcript.
178 Reads the most recently modified non-agent JSONL file in the project's
179 ~/.claude/projects/<dir>/ folder, sums ``cache_read_input_tokens``,
180 ``cache_creation_input_tokens``, and ``input_tokens`` across all unique
181 assistant entries (deduplicated by UUID to avoid double-counting), and
182 returns the aggregate hit rate.
184 Formula: hit_rate = cache_read / (cache_read + cache_write + uncached) * 100
185 """
186 project_folder = get_project_folder(cwd)
187 if project_folder is None:
188 return None
190 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
191 if not jsonl_files:
192 return None
194 latest = max(jsonl_files, key=lambda f: f.stat().st_mtime)
196 cache_read = 0
197 cache_write = 0
198 uncached = 0
199 seen_uuids: set[str] = set()
201 try:
202 with open(latest, encoding="utf-8") as f:
203 for line in f:
204 line = line.strip()
205 if not line:
206 continue
207 try:
208 record = json.loads(line)
209 except json.JSONDecodeError:
210 continue
211 if record.get("type") != "assistant":
212 continue
213 uuid = record.get("uuid")
214 if uuid:
215 if uuid in seen_uuids:
216 continue
217 seen_uuids.add(uuid)
218 usage = record.get("message", {}).get("usage", {})
219 if not usage:
220 continue
221 cache_read += int(usage.get("cache_read_input_tokens", 0))
222 cache_write += int(usage.get("cache_creation_input_tokens", 0))
223 uncached += int(usage.get("input_tokens", 0))
224 except OSError:
225 return None
227 total = cache_read + cache_write + uncached
228 if total == 0:
229 return None
231 return {
232 "cache_read": cache_read,
233 "cache_write": cache_write,
234 "uncached": uncached,
235 "hit_rate_pct": round(cache_read / total * 100),
236 }
239def _render(
240 summary: dict[str, Any],
241 logger: Logger,
242 skill_stats: dict[str, dict[str, int]] | None = None,
243 cache_rate: dict[str, Any] | None = None,
244) -> None:
245 """Print the savings report for an aggregated SQLite ``summary`` dict."""
246 total_processed = int(summary["total_in"]) + int(summary["total_out"])
247 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
248 saved = max(0, total_processed - in_context)
249 reduction = round(100 * saved / total_processed) if total_processed > 0 else 0
251 width = terminal_width()
252 bar_width = max(20, min(50, width - 30))
253 print(
254 f"Without savings: {_progress_bar(total_processed, total_processed, bar_width)} "
255 f"{_format_bytes(total_processed)} in conversation"
256 )
257 print(
258 f"With savings: {_progress_bar(in_context, total_processed, bar_width)} "
259 f"{_format_bytes(in_context)} in conversation"
260 )
261 print()
262 print(
263 f"{_format_bytes(saved)} processed by tools, never entered conversation. "
264 f"({reduction}% reduction)"
265 )
266 # Heuristic: ~100 bytes/sec of saved context ≈ time the user would have spent
267 # waiting for compaction or re-reading. The estimate is rough by design;
268 # FEAT-1625 may revisit once real telemetry is collected.
269 time_seconds = saved / 100.0 if saved > 0 else 0.0
270 print(f"{_time_gained(time_seconds)} session time gained.")
271 print()
273 per_tool: dict[str, dict[str, int]] = summary["per_tool"]
274 if per_tool:
275 ranked = sorted(per_tool.items(), key=lambda kv: kv[1]["bytes"], reverse=True)
276 for tool, stats in ranked:
277 print(
278 f" {tool:<13} {stats['calls']:>3} calls {_format_bytes(stats['bytes']):>10} used"
279 )
280 print()
282 cache_hits = int(summary["cache_hits"])
283 cache_bytes = int(summary["cache_bytes"])
284 if cache_hits:
285 print(f"Cache: {cache_hits} hits | {_format_bytes(cache_bytes)} saved")
286 else:
287 logger.info("Cache: no hits recorded in this session")
289 if cache_rate is not None:
290 cr = cache_rate["cache_read"]
291 cw = cache_rate["cache_write"]
292 u = cache_rate["uncached"]
293 pct = cache_rate["hit_rate_pct"]
294 print(f"Cache hit rate: {pct}% (cache_read={cr:,} | cache_write={cw:,} | uncached={u:,})")
296 if skill_stats:
297 print()
298 print("Skill health:")
299 ranked_skills = sorted(
300 skill_stats.items(), key=lambda kv: kv[1]["invocations"], reverse=True
301 )
302 for skill, counts in ranked_skills:
303 inv = counts["invocations"]
304 corr = counts["corrections"]
305 rate = round(100 * corr / inv) if inv > 0 else 0
306 print(f" {skill:<22} {inv:>3} invocations {corr:>2} corrections ({rate}%)")
307 elif skill_stats is not None:
308 logger.info("No skill events recorded yet.")
311def _render_fallback(state: dict[str, Any], logger: Logger) -> None:
312 """Render the ``.ll/ll-context-state.json`` fallback (token estimates)."""
313 estimated = int(state.get("estimated_tokens") or 0)
314 tool_calls = int(state.get("tool_calls") or 0)
315 breakdown = state.get("breakdown") or {}
317 logger.info(
318 "SQLite session store not found — falling back to .ll/ll-context-state.json "
319 "(enable analytics (analytics.enabled: true) and ensure analytics.capture.file_events is not disabled)."
320 )
321 print()
322 print(f"Estimated tokens in context: {estimated:,}")
323 print(f"Tool calls this session: {tool_calls}")
324 if isinstance(breakdown, dict) and breakdown:
325 print()
326 print("Per-tool token estimates:")
327 for tool, tokens in sorted(breakdown.items(), key=lambda kv: kv[1], reverse=True):
328 print(f" {str(tool):<20} {int(tokens):>8} tokens")
331def _print_json(
332 summary: dict[str, Any] | None,
333 state: dict[str, Any] | None,
334 skill_stats: dict[str, dict[str, int]] | None = None,
335 cache_rate: dict[str, Any] | None = None,
336) -> None:
337 """Emit a JSON document combining SQLite + fallback data."""
338 if summary is not None:
339 total_processed = int(summary["total_in"]) + int(summary["total_out"])
340 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
341 saved = max(0, total_processed - in_context)
342 skill_health = None
343 if skill_stats:
344 skill_health = [
345 {
346 "skill": skill,
347 "invocations": counts["invocations"],
348 "corrections": counts["corrections"],
349 "correction_rate": (
350 round(counts["corrections"] / counts["invocations"], 4)
351 if counts["invocations"] > 0
352 else 0.0
353 ),
354 }
355 for skill, counts in sorted(
356 skill_stats.items(), key=lambda kv: kv[1]["invocations"], reverse=True
357 )
358 ]
359 payload: dict[str, Any] = {
360 "source": "sqlite",
361 "bytes_processed": total_processed,
362 "bytes_in_context": in_context,
363 "bytes_saved": saved,
364 "reduction_pct": round(100 * saved / total_processed) if total_processed else 0,
365 "cache_hits": int(summary["cache_hits"]),
366 "cache_bytes_saved": int(summary["cache_bytes"]),
367 "cache_hit_rate_pct": cache_rate["hit_rate_pct"] if cache_rate else None,
368 "cache_read_tokens": cache_rate["cache_read"] if cache_rate else None,
369 "cache_write_tokens": cache_rate["cache_write"] if cache_rate else None,
370 "uncached_tokens": cache_rate["uncached"] if cache_rate else None,
371 "per_tool": summary["per_tool"],
372 "skill_health": skill_health,
373 }
374 elif state is not None:
375 payload = {
376 "source": "fallback",
377 "estimated_tokens": int(state.get("estimated_tokens") or 0),
378 "tool_calls": int(state.get("tool_calls") or 0),
379 "breakdown": state.get("breakdown") or {},
380 }
381 else:
382 payload = {"source": "none"}
383 print(json.dumps(payload, indent=2))
386def main_ctx_stats(argv: list[str] | None = None) -> int:
387 """Entry point for ll-ctx-stats command.
389 Read per-tool byte metrics from ``.ll/history.db`` (FEAT-1623) and print
390 a context-window savings summary. Falls back to
391 ``.ll/ll-context-state.json`` when the SQLite store is absent.
392 """
393 with cli_event_context(DEFAULT_DB_PATH, "ll-ctx-stats", sys.argv[1:]):
394 args = _parse_args(argv)
395 configure_output()
396 logger = Logger(use_color=use_color_enabled())
398 cwd = Path.cwd()
399 db_path = args.db if args.db is not None else cwd / DEFAULT_DB_RELPATH
400 state_path = cwd / DEFAULT_STATE_RELPATH
402 summary = _aggregate_tool_events(db_path)
403 skill_stats = _aggregate_skill_stats(db_path)
404 fallback = _load_fallback_state(state_path) if summary is None else None
405 cache_rate = _compute_cache_rate_from_jsonl(cwd)
407 if args.json_mode:
408 _print_json(summary, fallback, skill_stats, cache_rate)
409 return 0 if (summary is not None or fallback is not None) else 1
411 if summary is not None:
412 total_rows = int(summary["total_in"]) + int(summary["total_out"])
413 if total_rows == 0:
414 logger.warning(
415 "No analytic rows in .ll/history.db — enable analytics (analytics.enabled: true) "
416 "and ensure analytics.capture.file_events is not disabled, then run a few tool calls."
417 )
418 if fallback is None:
419 fallback = _load_fallback_state(state_path)
420 else:
421 _render(summary, logger, skill_stats, cache_rate)
422 return 0
424 if fallback is not None:
425 _render_fallback(fallback, logger)
426 return 0
428 logger.error(
429 "No context analytics found: neither .ll/history.db nor "
430 ".ll/ll-context-state.json contained data for this project."
431 )
432 return 1