Coverage for little_loops / cli / ctx_stats.py: 13%
268 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -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.config.features import LearningTestsConfig
28from little_loops.issue_parser import slugify
29from little_loops.learning_tests import list_records
30from little_loops.learning_tests.gate import is_record_stale
31from little_loops.learning_tests.import_scan import get_imported_packages
32from little_loops.logger import Logger
33from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
34from little_loops.user_messages import get_project_folder
36DEFAULT_DB_RELPATH = Path(".ll") / "history.db"
37DEFAULT_STATE_RELPATH = Path(".ll") / "ll-context-state.json"
40def _build_parser() -> argparse.ArgumentParser:
41 """Build the ll-ctx-stats argument parser (exposed for testing)."""
42 parser = argparse.ArgumentParser(
43 prog="ll-ctx-stats",
44 description=(
45 "Show context-window savings metrics and skill-health signals for the current project. "
46 "Reads per-tool byte metrics from .ll/history.db and renders how much data was processed "
47 "by tools vs. how much entered conversation context. Also surfaces per-skill invocation "
48 "frequency and correction rate from the same database."
49 ),
50 formatter_class=argparse.RawDescriptionHelpFormatter,
51 epilog="""
52Examples:
53 %(prog)s # Print savings summary and skill-health section
54 %(prog)s --db PATH # Use a non-default session database
55 %(prog)s --json # Output as JSON (includes skill_health array)
57Exit codes:
58 0 - Report rendered (data present or fallback used)
59 1 - No data found in either the SQLite store or the fallback file
60""",
61 )
62 parser.add_argument(
63 "--db",
64 type=Path,
65 default=None,
66 help="Path to the session database (default: .ll/history.db)",
67 )
68 parser.add_argument(
69 "-j",
70 "--json",
71 dest="json_mode",
72 action="store_true",
73 help="Output as JSON",
74 )
75 return parser
78def _parse_args(argv: list[str] | None) -> argparse.Namespace:
79 """Parse argv into a Namespace (exposed for testing)."""
80 return _build_parser().parse_args(argv)
83def _format_bytes(value: int) -> str:
84 """Render *value* bytes as a short ``KB``/``MB`` string."""
85 if value < 1024:
86 return f"{value} B"
87 if value < 1024 * 1024:
88 return f"{value / 1024:.1f} KB"
89 return f"{value / (1024 * 1024):.1f} MB"
92def _time_gained(seconds: float) -> str:
93 """Render *seconds* as a positive-tense ``+Xm`` string.
95 ``format_relative_time`` appends ``" ago"`` (it is designed for past-tense
96 durations); strip that suffix here so the line reads as savings rather
97 than elapsed time. The shared helper is intentionally left unchanged
98 (see Implementation Constraints #4 on FEAT-1624).
99 """
100 label = format_relative_time(seconds)
101 if label.endswith(" ago"):
102 label = label[: -len(" ago")]
103 return f"+{label}"
106def _progress_bar(value: int, ceiling: int, width: int) -> str:
107 """Return a ``|#### |`` bar of ``width`` columns scaled to ``value/ceiling``."""
108 if width < 3:
109 width = 3
110 inner = width - 2
111 if ceiling <= 0:
112 filled = 0
113 else:
114 filled = max(0, min(inner, round(inner * value / ceiling)))
115 return "|" + "#" * filled + " " * (inner - filled) + "|"
118def _aggregate_tool_events(db_path: Path) -> dict[str, Any] | None:
119 """Sum per-tool byte metrics from ``tool_events``.
121 Backfilled rows have ``NULL`` byte columns (see Implementation Constraints
122 #1 on FEAT-1624). Per-tool aggregation filters those rows out so historic
123 JSONL noise does not skew the summary; cache totals likewise.
125 Returns ``None`` when the database file is missing. Returns an empty
126 summary (all zeros) when the database exists but has no analytic rows.
127 """
128 if not db_path.exists():
129 return None
130 conn = sqlite3.connect(str(db_path))
131 try:
132 conn.row_factory = sqlite3.Row
133 try:
134 rows = conn.execute(
135 "SELECT tool_name, bytes_in, bytes_out, cache_hit "
136 "FROM tool_events WHERE bytes_in IS NOT NULL OR bytes_out IS NOT NULL"
137 ).fetchall()
138 except sqlite3.OperationalError:
139 return None
140 finally:
141 conn.close()
143 per_tool: dict[str, dict[str, int]] = defaultdict(lambda: {"calls": 0, "bytes": 0})
144 total_in = 0
145 total_out = 0
146 cache_hits = 0
147 cache_bytes = 0
148 for row in rows:
149 tool = (row["tool_name"] or "unknown").lower()
150 bin_ = int(row["bytes_in"] or 0)
151 bout = int(row["bytes_out"] or 0)
152 per_tool[tool]["calls"] += 1
153 per_tool[tool]["bytes"] += bout
154 total_in += bin_
155 total_out += bout
156 if row["cache_hit"]:
157 cache_hits += 1
158 cache_bytes += bout
160 return {
161 "total_in": total_in,
162 "total_out": total_out,
163 "cache_hits": cache_hits,
164 "cache_bytes": cache_bytes,
165 "per_tool": dict(per_tool),
166 }
169def _load_fallback_state(path: Path) -> dict[str, Any] | None:
170 """Return ``.ll/ll-context-state.json`` parsed, or ``None`` if absent/invalid."""
171 if not path.exists():
172 return None
173 try:
174 data = json.loads(path.read_text(encoding="utf-8"))
175 except (OSError, json.JSONDecodeError):
176 return None
177 return data if isinstance(data, dict) else None
180def _compute_cache_rate_from_jsonl(cwd: Path) -> dict[str, Any] | None:
181 """Compute session-aggregate cache hit rate from the most recent JSONL transcript.
183 Reads the most recently modified non-agent JSONL file in the project's
184 ~/.claude/projects/<dir>/ folder, sums ``cache_read_input_tokens``,
185 ``cache_creation_input_tokens``, and ``input_tokens`` across all unique
186 assistant entries (deduplicated by UUID to avoid double-counting), and
187 returns the aggregate hit rate.
189 Formula: hit_rate = cache_read / (cache_read + cache_write + uncached) * 100
190 """
191 project_folder = get_project_folder(cwd)
192 if project_folder is None:
193 return None
195 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
196 if not jsonl_files:
197 return None
199 latest = max(jsonl_files, key=lambda f: f.stat().st_mtime)
201 cache_read = 0
202 cache_write = 0
203 uncached = 0
204 seen_uuids: set[str] = set()
206 try:
207 with open(latest, encoding="utf-8") as f:
208 for line in f:
209 line = line.strip()
210 if not line:
211 continue
212 try:
213 record = json.loads(line)
214 except json.JSONDecodeError:
215 continue
216 if record.get("type") != "assistant":
217 continue
218 uuid = record.get("uuid")
219 if uuid:
220 if uuid in seen_uuids:
221 continue
222 seen_uuids.add(uuid)
223 usage = record.get("message", {}).get("usage", {})
224 if not usage:
225 continue
226 cache_read += int(usage.get("cache_read_input_tokens", 0))
227 cache_write += int(usage.get("cache_creation_input_tokens", 0))
228 uncached += int(usage.get("input_tokens", 0))
229 except OSError:
230 return None
232 total = cache_read + cache_write + uncached
233 if total == 0:
234 return None
236 return {
237 "cache_read": cache_read,
238 "cache_write": cache_write,
239 "uncached": uncached,
240 "hit_rate_pct": round(cache_read / total * 100),
241 }
244def _render(
245 summary: dict[str, Any],
246 logger: Logger,
247 skill_stats: dict[str, dict[str, int]] | None = None,
248 cache_rate: dict[str, Any] | None = None,
249 lt_stats: dict[str, Any] | None = None,
250) -> None:
251 """Print the savings report for an aggregated SQLite ``summary`` dict."""
252 total_processed = int(summary["total_in"]) + int(summary["total_out"])
253 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
254 saved = max(0, total_processed - in_context)
255 reduction = round(100 * saved / total_processed) if total_processed > 0 else 0
257 width = terminal_width()
258 bar_width = max(20, min(50, width - 30))
259 print(
260 f"Without savings: {_progress_bar(total_processed, total_processed, bar_width)} "
261 f"{_format_bytes(total_processed)} in conversation"
262 )
263 print(
264 f"With savings: {_progress_bar(in_context, total_processed, bar_width)} "
265 f"{_format_bytes(in_context)} in conversation"
266 )
267 print()
268 print(
269 f"{_format_bytes(saved)} processed by tools, never entered conversation. "
270 f"({reduction}% reduction)"
271 )
272 # Heuristic: ~100 bytes/sec of saved context ≈ time the user would have spent
273 # waiting for compaction or re-reading. The estimate is rough by design;
274 # FEAT-1625 may revisit once real telemetry is collected.
275 time_seconds = saved / 100.0 if saved > 0 else 0.0
276 print(f"{_time_gained(time_seconds)} session time gained.")
277 print()
279 per_tool: dict[str, dict[str, int]] = summary["per_tool"]
280 if per_tool:
281 ranked = sorted(per_tool.items(), key=lambda kv: kv[1]["bytes"], reverse=True)
282 for tool, stats in ranked:
283 print(
284 f" {tool:<13} {stats['calls']:>3} calls {_format_bytes(stats['bytes']):>10} used"
285 )
286 print()
288 cache_hits = int(summary["cache_hits"])
289 cache_bytes = int(summary["cache_bytes"])
290 if cache_hits:
291 print(f"Cache: {cache_hits} hits | {_format_bytes(cache_bytes)} saved")
292 else:
293 logger.info("Cache: no hits recorded in this session")
295 if cache_rate is not None:
296 cr = cache_rate["cache_read"]
297 cw = cache_rate["cache_write"]
298 u = cache_rate["uncached"]
299 pct = cache_rate["hit_rate_pct"]
300 print(f"Cache hit rate: {pct}% (cache_read={cr:,} | cache_write={cw:,} | uncached={u:,})")
302 if skill_stats:
303 print()
304 print("Skill health:")
305 ranked_skills = sorted(
306 skill_stats.items(), key=lambda kv: kv[1]["invocations"], reverse=True
307 )
308 for skill, counts in ranked_skills:
309 inv = counts["invocations"]
310 corr = counts["corrections"]
311 rate = round(100 * corr / inv) if inv > 0 else 0
312 print(f" {skill:<22} {inv:>3} invocations {corr:>2} corrections ({rate}%)")
313 elif skill_stats is not None:
314 logger.info("No skill events recorded yet.")
316 if lt_stats is not None:
317 _render_learning_tests_section(lt_stats)
320def _render_fallback(state: dict[str, Any], logger: Logger) -> None:
321 """Render the ``.ll/ll-context-state.json`` fallback (token estimates)."""
322 estimated = int(state.get("estimated_tokens") or 0)
323 tool_calls = int(state.get("tool_calls") or 0)
324 breakdown = state.get("breakdown") or {}
326 logger.info(
327 "SQLite session store not found — falling back to .ll/ll-context-state.json "
328 "(enable analytics (analytics.enabled: true) and ensure analytics.capture.file_events is not disabled)."
329 )
330 print()
331 print(f"Estimated tokens in context: {estimated:,}")
332 print(f"Tool calls this session: {tool_calls}")
333 if isinstance(breakdown, dict) and breakdown:
334 print()
335 print("Per-tool token estimates:")
336 for tool, tokens in sorted(breakdown.items(), key=lambda kv: kv[1], reverse=True):
337 print(f" {str(tool):<20} {int(tokens):>8} tokens")
340def _print_json(
341 summary: dict[str, Any] | None,
342 state: dict[str, Any] | None,
343 skill_stats: dict[str, dict[str, int]] | None = None,
344 cache_rate: dict[str, Any] | None = None,
345 lt_stats: dict[str, Any] | None = None,
346) -> None:
347 """Emit a JSON document combining SQLite + fallback data."""
348 if summary is not None:
349 total_processed = int(summary["total_in"]) + int(summary["total_out"])
350 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
351 saved = max(0, total_processed - in_context)
352 skill_health = None
353 if skill_stats:
354 skill_health = [
355 {
356 "skill": skill,
357 "invocations": counts["invocations"],
358 "corrections": counts["corrections"],
359 "correction_rate": (
360 round(counts["corrections"] / counts["invocations"], 4)
361 if counts["invocations"] > 0
362 else 0.0
363 ),
364 }
365 for skill, counts in sorted(
366 skill_stats.items(), key=lambda kv: kv[1]["invocations"], reverse=True
367 )
368 ]
369 payload: dict[str, Any] = {
370 "source": "sqlite",
371 "bytes_processed": total_processed,
372 "bytes_in_context": in_context,
373 "bytes_saved": saved,
374 "reduction_pct": round(100 * saved / total_processed) if total_processed else 0,
375 "cache_hits": int(summary["cache_hits"]),
376 "cache_bytes_saved": int(summary["cache_bytes"]),
377 "cache_hit_rate_pct": cache_rate["hit_rate_pct"] if cache_rate else None,
378 "cache_read_tokens": cache_rate["cache_read"] if cache_rate else None,
379 "cache_write_tokens": cache_rate["cache_write"] if cache_rate else None,
380 "uncached_tokens": cache_rate["uncached"] if cache_rate else None,
381 "per_tool": summary["per_tool"],
382 "skill_health": skill_health,
383 "learning_tests": lt_stats,
384 }
385 elif state is not None:
386 payload = {
387 "source": "fallback",
388 "estimated_tokens": int(state.get("estimated_tokens") or 0),
389 "tool_calls": int(state.get("tool_calls") or 0),
390 "breakdown": state.get("breakdown") or {},
391 "learning_tests": lt_stats,
392 }
393 else:
394 payload = {"source": "none"}
395 print(json.dumps(payload, indent=2))
398def _load_lt_config(cwd: Path) -> LearningTestsConfig:
399 """Load LearningTestsConfig from .ll/ll-config.json, defaulting to disabled."""
400 config_path = cwd / ".ll" / "ll-config.json"
401 if not config_path.exists():
402 return LearningTestsConfig()
403 try:
404 data = json.loads(config_path.read_text(encoding="utf-8"))
405 return LearningTestsConfig.from_dict(data.get("learning_tests", {}))
406 except (OSError, json.JSONDecodeError):
407 return LearningTestsConfig()
410def _compute_learning_tests_stats(
411 cwd: Path,
412 lt_config: LearningTestsConfig,
413) -> dict[str, Any] | None:
414 """Compute learning test registry stats.
416 Applies date-aware staleness reclassification: a record with status=proven
417 that exceeds stale_after_days is counted as stale, not proven (ENH-2208).
418 Returns None when learning_tests.enabled is False.
419 """
420 if not lt_config.enabled:
421 return None
423 records = list_records()
425 proven = 0
426 stale = 0
427 refuted = 0
428 last_date: str | None = None
429 known_slugs: set[str] = set()
431 for record in records:
432 if last_date is None or record.date > last_date:
433 last_date = record.date
434 known_slugs.add(slugify(record.target))
436 if record.status == "refuted":
437 refuted += 1
438 elif record.status == "stale" or (
439 record.status == "proven" and is_record_stale(record, lt_config.stale_after_days)
440 ):
441 stale += 1
442 else:
443 proven += 1
445 scan_dirs = [cwd / d for d in lt_config.scan_dirs]
446 imported = get_imported_packages(scan_dirs)
447 gaps = sorted(pkg for pkg in imported if slugify(pkg) not in known_slugs)
449 return {
450 "total": len(records),
451 "proven": proven,
452 "stale": stale,
453 "refuted": refuted,
454 "last_record": last_date,
455 "gaps": gaps,
456 }
459def _render_learning_tests_section(lt_stats: dict[str, Any]) -> None:
460 """Print the Learning Tests dashboard section."""
461 total = lt_stats["total"]
462 proven = lt_stats["proven"]
463 stale = lt_stats["stale"]
464 refuted = lt_stats["refuted"]
465 last_record = lt_stats["last_record"]
466 gaps: list[str] = lt_stats["gaps"]
468 print()
469 print("Learning tests:")
470 print(f" {total} total ({proven} proven, {stale} stale, {refuted} refuted)")
471 if last_record:
472 print(f" Last record: {last_record}")
473 if gaps:
474 print(f" Coverage gaps: {', '.join(gaps)}")
477def main_ctx_stats(argv: list[str] | None = None) -> int:
478 """Entry point for ll-ctx-stats command.
480 Read per-tool byte metrics from ``.ll/history.db`` (FEAT-1623) and print
481 a context-window savings summary. Falls back to
482 ``.ll/ll-context-state.json`` when the SQLite store is absent.
483 """
484 with cli_event_context(DEFAULT_DB_PATH, "ll-ctx-stats", sys.argv[1:]):
485 args = _parse_args(argv)
486 configure_output()
487 logger = Logger(use_color=use_color_enabled())
489 cwd = Path.cwd()
490 db_path = args.db if args.db is not None else cwd / DEFAULT_DB_RELPATH
491 state_path = cwd / DEFAULT_STATE_RELPATH
493 summary = _aggregate_tool_events(db_path)
494 skill_stats = _aggregate_skill_stats(db_path)
495 fallback = _load_fallback_state(state_path) if summary is None else None
496 cache_rate = _compute_cache_rate_from_jsonl(cwd)
497 lt_config = _load_lt_config(cwd)
498 lt_stats = _compute_learning_tests_stats(cwd, lt_config)
500 if args.json_mode:
501 _print_json(summary, fallback, skill_stats, cache_rate, lt_stats)
502 return 0 if (summary is not None or fallback is not None) else 1
504 if summary is not None:
505 total_rows = int(summary["total_in"]) + int(summary["total_out"])
506 if total_rows == 0:
507 logger.warning(
508 "No analytic rows in .ll/history.db — enable analytics (analytics.enabled: true) "
509 "and ensure analytics.capture.file_events is not disabled, then run a few tool calls."
510 )
511 if fallback is None:
512 fallback = _load_fallback_state(state_path)
513 else:
514 _render(summary, logger, skill_stats, cache_rate, lt_stats)
515 return 0
517 if fallback is not None:
518 _render_fallback(fallback, logger)
519 return 0
521 logger.error(
522 "No context analytics found: neither .ll/history.db nor "
523 ".ll/ll-context-state.json contained data for this project."
524 )
525 return 1