Coverage for little_loops / cli / ctx_stats.py: 16%
149 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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.output import (
21 configure_output,
22 format_relative_time,
23 terminal_width,
24 use_color_enabled,
25)
26from little_loops.logger import Logger
27from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
29DEFAULT_DB_RELPATH = Path(".ll") / "history.db"
30DEFAULT_STATE_RELPATH = Path(".ll") / "ll-context-state.json"
33def _build_parser() -> argparse.ArgumentParser:
34 """Build the ll-ctx-stats argument parser (exposed for testing)."""
35 parser = argparse.ArgumentParser(
36 prog="ll-ctx-stats",
37 description="Show context-window savings metrics for the current project",
38 formatter_class=argparse.RawDescriptionHelpFormatter,
39 epilog="""
40Examples:
41 %(prog)s # Print savings summary
42 %(prog)s --db PATH # Use a non-default session database
43 %(prog)s --json # Output as JSON
45Exit codes:
46 0 - Report rendered (data present or fallback used)
47 1 - No data found in either the SQLite store or the fallback file
48""",
49 )
50 parser.add_argument(
51 "--db",
52 type=Path,
53 default=None,
54 help="Path to the session database (default: .ll/history.db)",
55 )
56 parser.add_argument(
57 "-j",
58 "--json",
59 dest="json_mode",
60 action="store_true",
61 help="Output as JSON",
62 )
63 return parser
66def _parse_args(argv: list[str] | None) -> argparse.Namespace:
67 """Parse argv into a Namespace (exposed for testing)."""
68 return _build_parser().parse_args(argv)
71def _format_bytes(value: int) -> str:
72 """Render *value* bytes as a short ``KB``/``MB`` string."""
73 if value < 1024:
74 return f"{value} B"
75 if value < 1024 * 1024:
76 return f"{value / 1024:.1f} KB"
77 return f"{value / (1024 * 1024):.1f} MB"
80def _time_gained(seconds: float) -> str:
81 """Render *seconds* as a positive-tense ``+Xm`` string.
83 ``format_relative_time`` appends ``" ago"`` (it is designed for past-tense
84 durations); strip that suffix here so the line reads as savings rather
85 than elapsed time. The shared helper is intentionally left unchanged
86 (see Implementation Constraints #4 on FEAT-1624).
87 """
88 label = format_relative_time(seconds)
89 if label.endswith(" ago"):
90 label = label[: -len(" ago")]
91 return f"+{label}"
94def _progress_bar(value: int, ceiling: int, width: int) -> str:
95 """Return a ``|#### |`` bar of ``width`` columns scaled to ``value/ceiling``."""
96 if width < 3:
97 width = 3
98 inner = width - 2
99 if ceiling <= 0:
100 filled = 0
101 else:
102 filled = max(0, min(inner, round(inner * value / ceiling)))
103 return "|" + "#" * filled + " " * (inner - filled) + "|"
106def _aggregate_tool_events(db_path: Path) -> dict[str, Any] | None:
107 """Sum per-tool byte metrics from ``tool_events``.
109 Backfilled rows have ``NULL`` byte columns (see Implementation Constraints
110 #1 on FEAT-1624). Per-tool aggregation filters those rows out so historic
111 JSONL noise does not skew the summary; cache totals likewise.
113 Returns ``None`` when the database file is missing. Returns an empty
114 summary (all zeros) when the database exists but has no analytic rows.
115 """
116 if not db_path.exists():
117 return None
118 conn = sqlite3.connect(str(db_path))
119 try:
120 conn.row_factory = sqlite3.Row
121 try:
122 rows = conn.execute(
123 "SELECT tool_name, bytes_in, bytes_out, cache_hit "
124 "FROM tool_events WHERE bytes_in IS NOT NULL OR bytes_out IS NOT NULL"
125 ).fetchall()
126 except sqlite3.OperationalError:
127 return None
128 finally:
129 conn.close()
131 per_tool: dict[str, dict[str, int]] = defaultdict(lambda: {"calls": 0, "bytes": 0})
132 total_in = 0
133 total_out = 0
134 cache_hits = 0
135 cache_bytes = 0
136 for row in rows:
137 tool = (row["tool_name"] or "unknown").lower()
138 bin_ = int(row["bytes_in"] or 0)
139 bout = int(row["bytes_out"] or 0)
140 per_tool[tool]["calls"] += 1
141 per_tool[tool]["bytes"] += bout
142 total_in += bin_
143 total_out += bout
144 if row["cache_hit"]:
145 cache_hits += 1
146 cache_bytes += bout
148 return {
149 "total_in": total_in,
150 "total_out": total_out,
151 "cache_hits": cache_hits,
152 "cache_bytes": cache_bytes,
153 "per_tool": dict(per_tool),
154 }
157def _load_fallback_state(path: Path) -> dict[str, Any] | None:
158 """Return ``.ll/ll-context-state.json`` parsed, or ``None`` if absent/invalid."""
159 if not path.exists():
160 return None
161 try:
162 data = json.loads(path.read_text(encoding="utf-8"))
163 except (OSError, json.JSONDecodeError):
164 return None
165 return data if isinstance(data, dict) else None
168def _render(summary: dict[str, Any], logger: Logger) -> None:
169 """Print the savings report for an aggregated SQLite ``summary`` dict."""
170 total_processed = int(summary["total_in"]) + int(summary["total_out"])
171 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
172 saved = max(0, total_processed - in_context)
173 reduction = round(100 * saved / total_processed) if total_processed > 0 else 0
175 width = terminal_width()
176 bar_width = max(20, min(50, width - 30))
177 print(
178 f"Without savings: {_progress_bar(total_processed, total_processed, bar_width)} "
179 f"{_format_bytes(total_processed)} in conversation"
180 )
181 print(
182 f"With savings: {_progress_bar(in_context, total_processed, bar_width)} "
183 f"{_format_bytes(in_context)} in conversation"
184 )
185 print()
186 print(
187 f"{_format_bytes(saved)} processed by tools, never entered conversation. "
188 f"({reduction}% reduction)"
189 )
190 # Heuristic: ~100 bytes/sec of saved context ≈ time the user would have spent
191 # waiting for compaction or re-reading. The estimate is rough by design;
192 # FEAT-1625 may revisit once real telemetry is collected.
193 time_seconds = saved / 100.0 if saved > 0 else 0.0
194 print(f"{_time_gained(time_seconds)} session time gained.")
195 print()
197 per_tool: dict[str, dict[str, int]] = summary["per_tool"]
198 if per_tool:
199 ranked = sorted(per_tool.items(), key=lambda kv: kv[1]["bytes"], reverse=True)
200 for tool, stats in ranked:
201 print(
202 f" {tool:<13} {stats['calls']:>3} calls {_format_bytes(stats['bytes']):>10} used"
203 )
204 print()
206 cache_hits = int(summary["cache_hits"])
207 cache_bytes = int(summary["cache_bytes"])
208 if cache_hits:
209 print(f"Cache: {cache_hits} hits | {_format_bytes(cache_bytes)} saved")
210 else:
211 logger.info("Cache: no hits recorded in this session")
214def _render_fallback(state: dict[str, Any], logger: Logger) -> None:
215 """Render the ``.ll/ll-context-state.json`` fallback (token estimates)."""
216 estimated = int(state.get("estimated_tokens") or 0)
217 tool_calls = int(state.get("tool_calls") or 0)
218 breakdown = state.get("breakdown") or {}
220 logger.info(
221 "SQLite session store not found — falling back to .ll/ll-context-state.json "
222 "(enable analytics (analytics.enabled: true) and ensure analytics.capture.file_events is not disabled)."
223 )
224 print()
225 print(f"Estimated tokens in context: {estimated:,}")
226 print(f"Tool calls this session: {tool_calls}")
227 if isinstance(breakdown, dict) and breakdown:
228 print()
229 print("Per-tool token estimates:")
230 for tool, tokens in sorted(breakdown.items(), key=lambda kv: kv[1], reverse=True):
231 print(f" {str(tool):<20} {int(tokens):>8} tokens")
234def _print_json(summary: dict[str, Any] | None, state: dict[str, Any] | None) -> None:
235 """Emit a JSON document combining SQLite + fallback data."""
236 if summary is not None:
237 total_processed = int(summary["total_in"]) + int(summary["total_out"])
238 in_context = max(0, int(summary["total_out"]) - int(summary["cache_bytes"]))
239 saved = max(0, total_processed - in_context)
240 payload: dict[str, Any] = {
241 "source": "sqlite",
242 "bytes_processed": total_processed,
243 "bytes_in_context": in_context,
244 "bytes_saved": saved,
245 "reduction_pct": round(100 * saved / total_processed) if total_processed else 0,
246 "cache_hits": int(summary["cache_hits"]),
247 "cache_bytes_saved": int(summary["cache_bytes"]),
248 "per_tool": summary["per_tool"],
249 }
250 elif state is not None:
251 payload = {
252 "source": "fallback",
253 "estimated_tokens": int(state.get("estimated_tokens") or 0),
254 "tool_calls": int(state.get("tool_calls") or 0),
255 "breakdown": state.get("breakdown") or {},
256 }
257 else:
258 payload = {"source": "none"}
259 print(json.dumps(payload, indent=2))
262def main_ctx_stats(argv: list[str] | None = None) -> int:
263 """Entry point for ll-ctx-stats command.
265 Read per-tool byte metrics from ``.ll/history.db`` (FEAT-1623) and print
266 a context-window savings summary. Falls back to
267 ``.ll/ll-context-state.json`` when the SQLite store is absent.
268 """
269 with cli_event_context(DEFAULT_DB_PATH, "ll-ctx-stats", sys.argv[1:]):
270 args = _parse_args(argv)
271 configure_output()
272 logger = Logger(use_color=use_color_enabled())
274 cwd = Path.cwd()
275 db_path = args.db if args.db is not None else cwd / DEFAULT_DB_RELPATH
276 state_path = cwd / DEFAULT_STATE_RELPATH
278 summary = _aggregate_tool_events(db_path)
279 fallback = _load_fallback_state(state_path) if summary is None else None
281 if args.json_mode:
282 _print_json(summary, fallback)
283 return 0 if (summary is not None or fallback is not None) else 1
285 if summary is not None:
286 total_rows = int(summary["total_in"]) + int(summary["total_out"])
287 if total_rows == 0:
288 logger.warning(
289 "No analytic rows in .ll/history.db — enable analytics (analytics.enabled: true) "
290 "and ensure analytics.capture.file_events is not disabled, then run a few tool calls."
291 )
292 if fallback is None:
293 fallback = _load_fallback_state(state_path)
294 else:
295 _render(summary, logger)
296 return 0
298 if fallback is not None:
299 _render_fallback(fallback, logger)
300 return 0
302 logger.error(
303 "No context analytics found: neither .ll/history.db nor "
304 ".ll/ll-context-state.json contained data for this project."
305 )
306 return 1