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