feat: added read-selector analyzers and coverage plotting tools
- Updated read range expansion to use 1 leading and 3 trailing context lines. - Changed read.defaultLimit from 500 to 300 in settings defaults. - Updated read docs and tests to reflect the asymmetric context line behavior. - Added read-selector analyzers and replay simulators to evaluate coverage and savings. - Added plotting tools that output new session-stats PNG dashboards from local usage data.
This commit is contained in:
can1357
@@ -0,0 +1,525 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Per-session range-coverage analysis for read calls.
|
||||
|
||||
For each (session, file) we collect every read's [start, end] interval and
|
||||
build a coverage map. From that we derive:
|
||||
|
||||
* does the model follow up after the first read?
|
||||
* where does each follow-up land relative to the initial range?
|
||||
forward : extends past the initial end
|
||||
backward : extends before the initial start
|
||||
inside : fully inside the initial range
|
||||
gap-above : disjoint, above the initial range (start > init_end + 1)
|
||||
gap-below : disjoint, below the initial range (end < init_start - 1)
|
||||
* how many disjoint regions does the final coverage have?
|
||||
* how does total covered lines compare to the initial range?
|
||||
|
||||
`:raw` / `:conflicts` reads don't have line bounds; they're tracked in a
|
||||
separate cohort and excluded from interval math.
|
||||
|
||||
Outputs:
|
||||
scripts/session-stats/out/selector-coverage.png
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
|
||||
DB_PATH = Path.home() / ".omp" / "stats.db"
|
||||
OUT_DIR = Path(__file__).resolve().parent / "out"
|
||||
DEFAULT_SINCE = "2026-05-04"
|
||||
|
||||
# When a read has start but no explicit end (`:50` or default bare path with
|
||||
# no offset/limit), assume the read tool returns this many lines. The read
|
||||
# tool's default page is 500.
|
||||
DEFAULT_PAGE = 500
|
||||
|
||||
_RANGE_RE = re.compile(r"^(\d+)(?:([-+])(\d+))?$")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Selector parsing
|
||||
|
||||
def parse_selector(path: str) -> tuple[str, int | None, int | None, str]:
|
||||
"""Returns (base_path, start, end, kind)."""
|
||||
if not path:
|
||||
return path, None, None, "none"
|
||||
tail_idx = path.rfind("/")
|
||||
tail = path[tail_idx + 1 :]
|
||||
colon = tail.rfind(":")
|
||||
if colon < 0:
|
||||
return path, None, None, "none"
|
||||
suffix = tail[colon + 1 :]
|
||||
base = (path[: tail_idx + 1] + tail[:colon]) if tail_idx >= 0 else tail[:colon]
|
||||
if suffix == "raw":
|
||||
return base, None, None, "raw"
|
||||
if suffix == "conflicts":
|
||||
return base, None, None, "conflicts"
|
||||
m = _RANGE_RE.match(suffix)
|
||||
if not m:
|
||||
return path, None, None, "none"
|
||||
start = int(m.group(1))
|
||||
op = m.group(2)
|
||||
nval = m.group(3)
|
||||
if op == "-" and nval is not None:
|
||||
return base, start, int(nval), "range"
|
||||
if op == "+" and nval is not None:
|
||||
return base, start, start + int(nval) - 1, "range"
|
||||
# bare `:N` — open-ended; assume one page.
|
||||
return base, start, start + DEFAULT_PAGE - 1, "range"
|
||||
|
||||
|
||||
def args_to_interval(arg_json: str | None) -> tuple[str, int | None, int | None, str] | None:
|
||||
"""Decode arg_json into (base, start, end, kind)."""
|
||||
if not arg_json:
|
||||
return None
|
||||
try:
|
||||
obj = json.loads(arg_json)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
path = obj.get("path")
|
||||
if not isinstance(path, str):
|
||||
return None
|
||||
base, start, end, kind = parse_selector(path)
|
||||
if kind != "none":
|
||||
return base, start, end, kind
|
||||
# Legacy offset/limit.
|
||||
offset = obj.get("offset")
|
||||
limit = obj.get("limit")
|
||||
if isinstance(offset, int) and isinstance(limit, int) and offset >= 1 and limit >= 1:
|
||||
return path, offset, offset + limit - 1, "range"
|
||||
if isinstance(offset, int) and offset >= 1:
|
||||
return path, offset, offset + DEFAULT_PAGE - 1, "range"
|
||||
# Bare path — read tool default returns first page. Marked as `default`
|
||||
# so the rest of the script can exclude it from "explicit selector".
|
||||
return path, 1, DEFAULT_PAGE, "default"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Coverage math
|
||||
|
||||
def merge_intervals(ivs: list[tuple[int, int]]) -> list[tuple[int, int]]:
|
||||
"""Merge overlapping / adjacent intervals. Inclusive bounds."""
|
||||
if not ivs:
|
||||
return []
|
||||
ivs = sorted(ivs)
|
||||
out = [ivs[0]]
|
||||
for s, e in ivs[1:]:
|
||||
ls, le = out[-1]
|
||||
if s <= le + 1: # touching or overlapping
|
||||
out[-1] = (ls, max(le, e))
|
||||
else:
|
||||
out.append((s, e))
|
||||
return out
|
||||
|
||||
|
||||
def classify_followup(
|
||||
s: int, e: int, init_s: int, init_e: int
|
||||
) -> str:
|
||||
"""Where does follow-up [s,e] land relative to initial [init_s, init_e]?"""
|
||||
if s >= init_s and e <= init_e:
|
||||
return "inside"
|
||||
if s > init_e + 1:
|
||||
return "gap-above"
|
||||
if e < init_s - 1:
|
||||
return "gap-below"
|
||||
if e > init_e and s >= init_s:
|
||||
return "forward"
|
||||
if s < init_s and e <= init_e:
|
||||
return "backward"
|
||||
# Spans both sides of initial range.
|
||||
return "both"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Pull
|
||||
|
||||
def iter_reads(conn: sqlite3.Connection, since_ms: int):
|
||||
sql = """
|
||||
SELECT session_file, seq, timestamp, arg_json
|
||||
FROM ss_tool_calls
|
||||
WHERE tool_name = 'read' AND timestamp >= ?
|
||||
ORDER BY session_file, seq
|
||||
"""
|
||||
return conn.execute(sql, (since_ms,))
|
||||
|
||||
|
||||
def collect(conn, since_ms) -> dict[tuple[str, str], list[tuple[int, int, int, str]]]:
|
||||
"""key (session, file) -> ordered list of (seq, start, end, kind)."""
|
||||
by_key: dict[tuple[str, str], list[tuple[int, int | None, int | None, str]]] = defaultdict(list)
|
||||
for session, seq, _ts, arg_json in iter_reads(conn, since_ms):
|
||||
parsed = args_to_interval(arg_json)
|
||||
if parsed is None:
|
||||
continue
|
||||
base, start, end, kind = parsed
|
||||
if not base or base.endswith("/") or "://" in base:
|
||||
continue
|
||||
by_key[(session, base)].append((seq, start, end, kind))
|
||||
return by_key
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Analyze
|
||||
|
||||
def analyze(by_key: dict) -> dict:
|
||||
"""Compute coverage statistics over (session, file) groups whose FIRST
|
||||
read is a numeric range."""
|
||||
eligible: list[dict] = []
|
||||
first_kind_counts: Counter = Counter()
|
||||
# Position counter across all follow-ups (not just first one).
|
||||
followup_positions: Counter = Counter()
|
||||
for (session, base), reads in by_key.items():
|
||||
first = reads[0]
|
||||
_, s0, e0, kind0 = first
|
||||
first_kind_counts[kind0] += 1
|
||||
if kind0 != "range" or s0 is None or e0 is None:
|
||||
continue
|
||||
followups = reads[1:]
|
||||
|
||||
intervals = [(s0, e0)]
|
||||
followup_kinds: list[str] = []
|
||||
first_followup_pos: str | None = None
|
||||
|
||||
for _, s, e, k in followups:
|
||||
followup_kinds.append(k)
|
||||
# `range` (explicit) and `default` (bare path → first page) both
|
||||
# contribute a known interval to coverage. `raw`/`conflicts`
|
||||
# have no line bounds so we skip them here.
|
||||
if k not in ("range", "default") or s is None or e is None:
|
||||
continue
|
||||
pos = classify_followup(s, e, s0, e0)
|
||||
followup_positions[pos] += 1
|
||||
if first_followup_pos is None:
|
||||
first_followup_pos = pos
|
||||
intervals.append((s, e))
|
||||
|
||||
merged = merge_intervals(intervals)
|
||||
covered_lines = sum(e - s + 1 for s, e in merged)
|
||||
init_size = e0 - s0 + 1
|
||||
regions = len(merged)
|
||||
# Span = bounding box length, gaps = span - covered.
|
||||
bbox = (merged[0][0], merged[-1][1])
|
||||
span = bbox[1] - bbox[0] + 1
|
||||
gap_lines = span - covered_lines
|
||||
extra_lines = max(0, covered_lines - init_size) # new lines past initial
|
||||
|
||||
eligible.append({
|
||||
"session": session,
|
||||
"file": base,
|
||||
"init_start": s0,
|
||||
"init_end": e0,
|
||||
"init_size": init_size,
|
||||
"n_followups": len(followups),
|
||||
"n_range_followups": sum(1 for k in followup_kinds if k in ("range", "default")),
|
||||
"n_raw_followups": sum(1 for k in followup_kinds if k == "raw"),
|
||||
"first_followup_pos": first_followup_pos,
|
||||
"intervals": merged,
|
||||
"regions": regions,
|
||||
"covered": covered_lines,
|
||||
"extra_lines": extra_lines,
|
||||
"span": span,
|
||||
"gap_lines": gap_lines,
|
||||
})
|
||||
return {
|
||||
"eligible": eligible,
|
||||
"first_kind": dict(first_kind_counts),
|
||||
"followup_pos": dict(followup_positions),
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Report
|
||||
|
||||
POS_ORDER = ["forward", "backward", "inside", "both", "gap-above", "gap-below"]
|
||||
POS_COLORS = {
|
||||
"forward": "#2563eb",
|
||||
"backward": "#0f766e",
|
||||
"inside": "#9ca3af",
|
||||
"both": "#7c3aed",
|
||||
"gap-above": "#dc2626",
|
||||
"gap-below": "#d97706",
|
||||
}
|
||||
POS_HELP = {
|
||||
"forward": "extended past initial end",
|
||||
"backward": "extended before initial start",
|
||||
"inside": "re-read inside the initial range",
|
||||
"both": "extended on both sides",
|
||||
"gap-above": "disjoint hop above initial",
|
||||
"gap-below": "disjoint hop below initial",
|
||||
}
|
||||
|
||||
|
||||
def report(stats: dict) -> None:
|
||||
eligible = stats["eligible"]
|
||||
first_kind = stats["first_kind"]
|
||||
total_pairs = sum(first_kind.values())
|
||||
|
||||
print("first-read selector breakdown across (session, file) pairs:")
|
||||
for k in ("default", "range", "raw", "conflicts", "none"):
|
||||
n = first_kind.get(k, 0)
|
||||
if n == 0:
|
||||
continue
|
||||
print(f" {k:<10} {n:>8,} {100*n/total_pairs:>5.1f}%")
|
||||
print(f" total {total_pairs:>8,}")
|
||||
|
||||
if not eligible:
|
||||
print("\nno (session, file) pairs with a ranged first read.")
|
||||
return
|
||||
|
||||
n = len(eligible)
|
||||
with_followup = sum(1 for e in eligible if e["n_followups"] > 0)
|
||||
with_range_followup = sum(1 for e in eligible if e["n_range_followups"] > 0)
|
||||
print(f"\nfor {n:,} (session, file) pairs whose first read was a range:")
|
||||
print(f" any follow-up read : {with_followup:>8,} ({100*with_followup/n:.1f}%)")
|
||||
print(f" follow-up with a range : {with_range_followup:>8,} ({100*with_range_followup/n:.1f}%)")
|
||||
print()
|
||||
print(" ----- follow-up position breakdown (all follow-up reads) -----")
|
||||
positions = stats["followup_pos"]
|
||||
total_pos = sum(positions.values())
|
||||
for k in POS_ORDER:
|
||||
v = positions.get(k, 0)
|
||||
if v == 0:
|
||||
continue
|
||||
print(f" {k:<10} {v:>8,} ({100*v/total_pos:>5.1f}%) -- {POS_HELP[k]}")
|
||||
|
||||
# Region count distribution.
|
||||
regions = np.array([e["regions"] for e in eligible], dtype=np.int64)
|
||||
print(f"\ndisjoint regions in final coverage (per session/file):")
|
||||
print(f" mean={regions.mean():.2f} median={int(np.median(regions))} "
|
||||
f"p90={int(np.percentile(regions,90))} max={int(regions.max())}")
|
||||
edges = [1, 2, 3, 4, 6, 11, 10**6]
|
||||
labels = ["1 (contig)", "2", "3", "4-5", "6-10", "11+"]
|
||||
hist, _ = np.histogram(regions, bins=edges)
|
||||
for label, nb in zip(labels, hist):
|
||||
print(f" {label:<10} {nb:>8,} ({100*nb/regions.size:>5.1f}%)")
|
||||
|
||||
# Extra lines vs initial (only when follow-ups exist).
|
||||
fu = [e for e in eligible if e["n_range_followups"] > 0]
|
||||
extra = np.array([e["extra_lines"] for e in fu], dtype=np.int64)
|
||||
if extra.size:
|
||||
print(f"\nextra lines covered beyond initial range (n={extra.size:,}):")
|
||||
print(f" mean={extra.mean():.0f} median={int(np.median(extra))} "
|
||||
f"p75={int(np.percentile(extra,75))} p90={int(np.percentile(extra,90))} "
|
||||
f"max={int(extra.max())}")
|
||||
edges = [0, 1, 51, 201, 501, 2001, 10**9]
|
||||
labels = ["0 (no new)", "1-50", "51-200", "201-500", "501-2000", "2000+"]
|
||||
hist, _ = np.histogram(extra, bins=edges)
|
||||
for label, nb in zip(labels, hist):
|
||||
print(f" {label:<12} {nb:>8,} ({100*nb/extra.size:>5.1f}%)")
|
||||
|
||||
# Coverage ratio.
|
||||
init_sizes = np.array([e["init_size"] for e in fu], dtype=np.int64)
|
||||
covered = np.array([e["covered"] for e in fu], dtype=np.int64)
|
||||
if init_sizes.size:
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
ratio = np.where(init_sizes > 0, covered / init_sizes, np.nan)
|
||||
ratio = ratio[np.isfinite(ratio)]
|
||||
print(f"\ntotal covered / initial size:")
|
||||
print(f" mean={ratio.mean():.2f}x median={np.median(ratio):.2f}x "
|
||||
f"p90={np.percentile(ratio,90):.2f}x")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Plot
|
||||
|
||||
def plot(stats: dict, since: str) -> Path | None:
|
||||
eligible = stats["eligible"]
|
||||
if not eligible:
|
||||
return None
|
||||
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
plt.rcParams.update({"figure.dpi": 110, "font.size": 10})
|
||||
|
||||
fig, axes = plt.subplots(2, 2, figsize=(15, 9))
|
||||
|
||||
# 1) Follow-up position breakdown.
|
||||
ax = axes[0, 0]
|
||||
pos = stats["followup_pos"]
|
||||
total = sum(pos.values())
|
||||
if total:
|
||||
keys = [k for k in POS_ORDER if pos.get(k, 0)]
|
||||
vals = [100 * pos[k] / total for k in keys]
|
||||
colors = [POS_COLORS[k] for k in keys]
|
||||
bars = ax.bar(keys, vals, color=colors, edgecolor="#111", linewidth=0.5)
|
||||
for bar, v, k in zip(bars, vals, keys):
|
||||
ax.text(bar.get_x() + bar.get_width() / 2, v + 1.0,
|
||||
f"{v:.1f}%\nn={pos[k]:,}", ha="center", va="bottom", fontsize=8)
|
||||
ax.set_title(
|
||||
f"where do follow-up reads land vs initial range (n={total:,})"
|
||||
)
|
||||
ax.set_ylabel("share of follow-up reads")
|
||||
ax.set_ylim(0, max(vals) + 12)
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda v, _: f"{v:.0f}%"))
|
||||
ax.grid(True, axis="y", alpha=0.25, linestyle="--")
|
||||
|
||||
# 2) Disjoint region count.
|
||||
ax = axes[0, 1]
|
||||
regions = np.array([e["regions"] for e in eligible], dtype=np.int64)
|
||||
edges = [1, 2, 3, 4, 6, 11, regions.max() + 1 if regions.size else 12]
|
||||
labels = ["1\n(contiguous)", "2", "3", "4-5", "6-10", "11+"]
|
||||
hist, _ = np.histogram(regions, bins=edges)
|
||||
pct = 100 * hist / regions.size
|
||||
colors = ["#16a34a"] + ["#2563eb"] * 5
|
||||
bars = ax.bar(labels, pct, color=colors, edgecolor="#111", linewidth=0.5)
|
||||
for bar, p, h in zip(bars, pct, hist):
|
||||
ax.text(bar.get_x() + bar.get_width() / 2, p + 1.5,
|
||||
f"{p:.1f}%\nn={h:,}", ha="center", va="bottom", fontsize=8)
|
||||
ax.set_title(f"disjoint regions in final coverage (n={regions.size:,})")
|
||||
ax.set_ylabel("share of (session, file) pairs")
|
||||
ax.set_ylim(0, max(pct.max() + 12, 20))
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda v, _: f"{v:.0f}%"))
|
||||
ax.grid(True, axis="y", alpha=0.25, linestyle="--")
|
||||
|
||||
# 3) Extra lines past initial.
|
||||
ax = axes[1, 0]
|
||||
fu = [e for e in eligible if e["n_range_followups"] > 0]
|
||||
extra = np.array([e["extra_lines"] for e in fu], dtype=np.int64)
|
||||
if extra.size:
|
||||
edges = [0, 1, 51, 201, 501, 2001, max(extra.max(), 2001) + 1]
|
||||
labels = ["0", "1-50", "51-200", "201-500", "501-2000", "2000+"]
|
||||
hist, _ = np.histogram(extra, bins=edges)
|
||||
pct = 100 * hist / extra.size
|
||||
bars = ax.bar(labels, pct, color="#d97706", edgecolor="#7c2d12", linewidth=0.5)
|
||||
for bar, p, h in zip(bars, pct, hist):
|
||||
ax.text(bar.get_x() + bar.get_width() / 2, p + 1.5,
|
||||
f"{p:.1f}%\nn={h:,}", ha="center", va="bottom", fontsize=8)
|
||||
ax.set_title(
|
||||
f"extra lines covered beyond initial range\n"
|
||||
f"(only pairs that follow up, n={extra.size:,})"
|
||||
)
|
||||
ax.set_ylabel("share of pairs")
|
||||
ax.set_ylim(0, max(pct.max() + 12, 20))
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda v, _: f"{v:.0f}%"))
|
||||
ax.grid(True, axis="y", alpha=0.25, linestyle="--")
|
||||
|
||||
# 4) Coverage ratio CDF.
|
||||
ax = axes[1, 1]
|
||||
init_sizes = np.array([e["init_size"] for e in fu], dtype=np.int64)
|
||||
covered = np.array([e["covered"] for e in fu], dtype=np.int64)
|
||||
if init_sizes.size:
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
ratio = np.where(init_sizes > 0, covered / init_sizes, np.nan)
|
||||
ratio = ratio[np.isfinite(ratio)]
|
||||
ratio.sort()
|
||||
cdf = np.arange(1, ratio.size + 1) / ratio.size
|
||||
ax.plot(ratio, cdf, color="#0f766e", linewidth=1.9)
|
||||
ax.axvline(1.0, color="#9ca3af", linestyle="--", linewidth=1.0,
|
||||
label="covered = initial")
|
||||
for q in (0.5, 0.9):
|
||||
x = np.interp(q, cdf, ratio)
|
||||
ax.scatter([x], [q], color="#dc2626", s=22, zorder=3)
|
||||
ax.annotate(f"p{int(q*100)}={x:.2f}x", (x, q),
|
||||
textcoords="offset points", xytext=(6, -8), fontsize=9)
|
||||
ax.set_xscale("log")
|
||||
ax.set_xlim(0.8, max(ratio.max(), 10))
|
||||
ax.set_xlabel("total covered lines / initial range (×, log)")
|
||||
ax.set_ylabel("CDF of pairs with follow-up")
|
||||
ax.set_title(f"how much of the file does the session end up reading? (n={ratio.size:,})")
|
||||
ax.set_ylim(0, 1.01)
|
||||
ax.legend(loc="lower right", frameon=False)
|
||||
ax.grid(True, which="both", alpha=0.25, linestyle="--")
|
||||
|
||||
fig.suptitle(f"selector reads — coverage map analysis (since {since})", fontsize=13, y=0.995)
|
||||
fig.tight_layout()
|
||||
p = OUT_DIR / "selector-coverage.png"
|
||||
fig.savefig(p, bbox_inches="tight")
|
||||
plt.close(fig)
|
||||
return p
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Examples
|
||||
|
||||
def dump_examples(stats: dict, k: int = 8) -> None:
|
||||
"""Print a few coverage-map examples for sanity / intuition."""
|
||||
fu = [e for e in stats["eligible"] if e["n_range_followups"] > 0]
|
||||
if not fu:
|
||||
return
|
||||
# Bucket by region count → pick one example from each bucket; for buckets
|
||||
# with many candidates prefer one whose initial range isn't the bare-path
|
||||
# default [1, 500] so the maps look more meaningful.
|
||||
buckets: dict[int, list[dict]] = defaultdict(list)
|
||||
for e in fu:
|
||||
bucket = min(e["regions"], 10)
|
||||
buckets[bucket].append(e)
|
||||
picks: list[dict] = []
|
||||
for r in sorted(buckets.keys()):
|
||||
candidates = buckets[r]
|
||||
# Prefer ones with non-default initial windows.
|
||||
non_default = [c for c in candidates if (c["init_start"], c["init_end"]) != (1, DEFAULT_PAGE)]
|
||||
chosen = non_default[0] if non_default else candidates[0]
|
||||
picks.append(chosen)
|
||||
if len(picks) >= k:
|
||||
break
|
||||
print("\nexample coverage maps:")
|
||||
for e in picks[:k]:
|
||||
# Compact ASCII map of intervals over the bounding range.
|
||||
bbox_lo = e["intervals"][0][0]
|
||||
bbox_hi = e["intervals"][-1][1]
|
||||
width = 50
|
||||
span = max(1, bbox_hi - bbox_lo)
|
||||
bar = ["·"] * width
|
||||
for s, ee in e["intervals"]:
|
||||
i0 = int((s - bbox_lo) / span * (width - 1))
|
||||
i1 = int((ee - bbox_lo) / span * (width - 1))
|
||||
for i in range(i0, i1 + 1):
|
||||
bar[i] = "█"
|
||||
# Highlight initial range positions.
|
||||
init_s, init_e = e["init_start"], e["init_end"]
|
||||
i0 = int((init_s - bbox_lo) / span * (width - 1))
|
||||
i1 = int((init_e - bbox_lo) / span * (width - 1))
|
||||
for i in range(i0, i1 + 1):
|
||||
if bar[i] == "█":
|
||||
bar[i] = "▓"
|
||||
bar_str = "".join(bar)
|
||||
file_short = e["file"][-50:]
|
||||
print(f" [{bar_str}] regions={e['regions']:>2} "
|
||||
f"init=[{e['init_start']},{e['init_end']}] "
|
||||
f"covered={e['covered']:>4} span={e['span']:>4} {file_short}")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Entry
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
|
||||
ap.add_argument("--since", default=DEFAULT_SINCE,
|
||||
help=f"only reads at or after this date (default {DEFAULT_SINCE})")
|
||||
ap.add_argument("--examples", type=int, default=8,
|
||||
help="how many coverage-map examples to print (default 8)")
|
||||
args = ap.parse_args()
|
||||
|
||||
since = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
||||
since_ms = int(since.timestamp() * 1000)
|
||||
|
||||
if not DB_PATH.exists():
|
||||
sys.exit(f"db missing: {DB_PATH}")
|
||||
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
|
||||
by_key = collect(conn, since_ms)
|
||||
conn.close()
|
||||
|
||||
print(f"loaded {sum(len(v) for v in by_key.values()):,} read calls across "
|
||||
f"{len(by_key):,} (session, file) pairs since {args.since}")
|
||||
|
||||
stats = analyze(by_key)
|
||||
report(stats)
|
||||
if args.examples:
|
||||
dump_examples(stats, args.examples)
|
||||
out = plot(stats, args.since)
|
||||
if out:
|
||||
print(f"\nwrote {out}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,726 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Replay-based optimizer for the read tool's config.
|
||||
|
||||
Inputs (from ~/.omp/stats.db, since --since):
|
||||
* every `read` call's args (selector / offset / limit / bare)
|
||||
* every result's `[Showing lines A-B of N]` footer → actual returned
|
||||
range AND the file's total line count
|
||||
* every result's `[Output truncated` marker → byte cap hit
|
||||
|
||||
For each (session, file) pair we replay the sequence under a candidate
|
||||
config C = (default_page D, line cap L, byte cap B*) and add up:
|
||||
reads kept * estimated_tokens(range, file) + reads kept * ROUNDTRIP
|
||||
|
||||
`*` byte cap is modelled as "an explicit range read of size > B/avg_bpl is
|
||||
clipped to floor(B/avg_bpl) lines" so we don't have to know raw bytes.
|
||||
|
||||
We sweep D and L over a grid, find the (D, L) minimizing simulated total
|
||||
tokens, and verify the simulator reproduces the baseline within ~5% of the
|
||||
actually-observed spend.
|
||||
|
||||
Output:
|
||||
scripts/session-stats/out/read-config-sweep.png
|
||||
console table with the recommended config + savings
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import NamedTuple
|
||||
|
||||
import matplotlib.pyplot as plt
|
||||
import matplotlib.patches as mpatches
|
||||
import numpy as np
|
||||
|
||||
DB_PATH = Path.home() / ".omp" / "stats.db"
|
||||
OUT_DIR = Path(__file__).resolve().parent / "out"
|
||||
DEFAULT_SINCE = "2026-05-04"
|
||||
|
||||
# Tool current values (packages/coding-agent/src/session/streaming-output.ts +
|
||||
# tools/read.ts) — used for baseline comparison.
|
||||
CURRENT_DEFAULT = 3000
|
||||
CURRENT_LINE_CAP = 3000
|
||||
CURRENT_BYTE_CAP = 50 * 1024 # 50 KB
|
||||
|
||||
# Average bytes per line — only used to convert byte cap → line cap when the
|
||||
# model didn't pass an explicit limit. Computed at runtime from observed
|
||||
# bytes_per_line per file, with this as a fallback for files we never saw.
|
||||
FALLBACK_BPL = 60.0
|
||||
FALLBACK_TPL = 12.0 # tokens per line if a file has no observed reads
|
||||
|
||||
# Cost of an extra tool roundtrip: at minimum the assistant text+thinking
|
||||
# preceding the call (median ~120-250 tokens) + the call envelope + the
|
||||
# result-header overhead. We charge a flat 200 tokens per call kept; the
|
||||
# answer is qualitatively stable across 50-400.
|
||||
ROUNDTRIP_OVERHEAD = 200
|
||||
|
||||
# Selector parser (reuses the same rules as analyze_selector_reads.py).
|
||||
_RANGE_RE = re.compile(r"^(\d+)(?:([-+])(\d+))?$")
|
||||
_FOOTER_RE = re.compile(
|
||||
r"\[Showing lines (\d+)-(\d+) of (\d+)\."
|
||||
)
|
||||
_TRUNCATED_RE = re.compile(r"\[Output truncated")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Selector → intent
|
||||
|
||||
class Intent(NamedTuple):
|
||||
kind: str # 'bare' | 'range' | 'raw' | 'conflicts' | 'other'
|
||||
start: int | None # requested start line (1-indexed) — only meaningful for 'range'
|
||||
end: int | None # requested end line (1-indexed, inclusive) — None = open-ended
|
||||
|
||||
|
||||
def parse_selector(path: str) -> tuple[str, Intent]:
|
||||
tail_idx = path.rfind("/")
|
||||
tail = path[tail_idx + 1 :]
|
||||
colon = tail.rfind(":")
|
||||
if colon < 0:
|
||||
return path, Intent("bare", None, None)
|
||||
suffix = tail[colon + 1 :]
|
||||
base = (path[: tail_idx + 1] + tail[:colon]) if tail_idx >= 0 else tail[:colon]
|
||||
if suffix == "raw":
|
||||
return base, Intent("raw", None, None)
|
||||
if suffix == "conflicts":
|
||||
return base, Intent("conflicts", None, None)
|
||||
m = _RANGE_RE.match(suffix)
|
||||
if not m:
|
||||
return path, Intent("other", None, None)
|
||||
s = int(m.group(1))
|
||||
op, nval = m.group(2), m.group(3)
|
||||
if op == "-" and nval is not None:
|
||||
return base, Intent("range", s, int(nval))
|
||||
if op == "+" and nval is not None:
|
||||
return base, Intent("range", s, s + int(nval) - 1)
|
||||
return base, Intent("range", s, None) # open-ended `:N`
|
||||
|
||||
|
||||
def parse_args(arg_json: str | None) -> tuple[str | None, Intent]:
|
||||
if not arg_json:
|
||||
return None, Intent("other", None, None)
|
||||
try:
|
||||
obj = json.loads(arg_json)
|
||||
except json.JSONDecodeError:
|
||||
return None, Intent("other", None, None)
|
||||
path = obj.get("path")
|
||||
if not isinstance(path, str):
|
||||
return None, Intent("other", None, None)
|
||||
base, intent = parse_selector(path)
|
||||
if intent.kind != "bare":
|
||||
return base, intent
|
||||
# Legacy offset/limit treated as an explicit range.
|
||||
offset = obj.get("offset")
|
||||
limit = obj.get("limit")
|
||||
if isinstance(offset, int) and isinstance(limit, int) and offset >= 1 and limit >= 1:
|
||||
return path, Intent("range", offset, offset + limit - 1)
|
||||
if isinstance(offset, int) and offset >= 1:
|
||||
return path, Intent("range", offset, None)
|
||||
return path, Intent("bare", None, None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Footer parser → (returned_start, returned_end, file_total_lines)
|
||||
|
||||
def parse_footer(tail: str | None) -> tuple[int | None, int | None, int | None, bool]:
|
||||
"""Returns (returned_a, returned_b, file_total, was_byte_truncated)."""
|
||||
if not tail:
|
||||
return None, None, None, False
|
||||
m = _FOOTER_RE.search(tail)
|
||||
if not m:
|
||||
return None, None, None, bool(_TRUNCATED_RE.search(tail))
|
||||
return (int(m.group(1)), int(m.group(2)), int(m.group(3)),
|
||||
bool(_TRUNCATED_RE.search(tail)))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Coverage utilities
|
||||
|
||||
def merge(ivs: list[tuple[int, int]]) -> list[tuple[int, int]]:
|
||||
if not ivs:
|
||||
return []
|
||||
ivs = sorted(ivs)
|
||||
out = [ivs[0]]
|
||||
for s, e in ivs[1:]:
|
||||
ls, le = out[-1]
|
||||
if s <= le + 1:
|
||||
out[-1] = (ls, max(le, e))
|
||||
else:
|
||||
out.append((s, e))
|
||||
return out
|
||||
|
||||
|
||||
def contained(s: int, e: int, ivs: list[tuple[int, int]]) -> bool:
|
||||
for ls, le in ivs:
|
||||
if ls <= s and le >= e:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def subtract(s: int, e: int, ivs: list[tuple[int, int]]) -> list[tuple[int, int]]:
|
||||
"""Return [s,e] minus the union of `ivs` as a list of remaining intervals."""
|
||||
out = [(s, e)]
|
||||
for ls, le in ivs:
|
||||
new = []
|
||||
for a, b in out:
|
||||
if le < a or ls > b:
|
||||
new.append((a, b))
|
||||
continue
|
||||
if ls > a:
|
||||
new.append((a, ls - 1))
|
||||
if le < b:
|
||||
new.append((le + 1, b))
|
||||
out = new
|
||||
if not out:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Data model
|
||||
|
||||
class ReadCall(NamedTuple):
|
||||
seq: int
|
||||
intent: Intent
|
||||
base: str
|
||||
actual_a: int | None # what came back: lines [actual_a, actual_b]
|
||||
actual_b: int | None
|
||||
file_total: int | None # from footer
|
||||
tokens: int # observed result tokens
|
||||
was_truncated: bool # [Output truncated marker present
|
||||
|
||||
|
||||
def fetch_reads(conn: sqlite3.Connection, since_ms: int) -> list[tuple[str, ReadCall]]:
|
||||
"""Returns list of (session, ReadCall) in (session, seq) order."""
|
||||
# Pull only the last 320 bytes of result_text — enough for the footer +
|
||||
# truncation marker, keeps the working set small.
|
||||
sql = """
|
||||
SELECT c.session_file,
|
||||
c.seq,
|
||||
c.arg_json,
|
||||
COALESCE(r.result_tokens, 0) AS tokens,
|
||||
substr(COALESCE(r.result_text, ''),
|
||||
MAX(1, LENGTH(COALESCE(r.result_text, '')) - 320))
|
||||
AS tail
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
WHERE c.tool_name = 'read' AND c.timestamp >= ?
|
||||
ORDER BY c.session_file, c.seq
|
||||
"""
|
||||
out: list[tuple[str, ReadCall]] = []
|
||||
for session, seq, arg_json, tokens, tail in conn.execute(sql, (since_ms,)):
|
||||
base, intent = parse_args(arg_json)
|
||||
if not base or base.endswith("/") or "://" in base:
|
||||
continue
|
||||
actual_a, actual_b, file_total, was_trunc = parse_footer(tail)
|
||||
out.append((session, ReadCall(seq, intent, base, actual_a, actual_b,
|
||||
file_total, int(tokens), was_trunc)))
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Per-file aggregates
|
||||
|
||||
class FileStats(NamedTuple):
|
||||
size_lines: int # best-effort estimate
|
||||
tokens_per_line: float
|
||||
bytes_per_line: float # only when we can derive (currently we can't, so fallback)
|
||||
|
||||
|
||||
def aggregate_files(reads: list[tuple[str, ReadCall]]) -> dict[str, FileStats]:
|
||||
"""Estimate per-file size + tokens/line from observed reads.
|
||||
|
||||
size_lines:
|
||||
- if any footer reported `of N`, use the max N seen for this file (most
|
||||
reads agree but files grow over time).
|
||||
- else fall back to max(actual_b or intent.end) ever observed.
|
||||
tokens_per_line:
|
||||
- weighted average of (tokens / lines_returned) across reads of this file
|
||||
that have a footer (so we know `lines_returned`).
|
||||
"""
|
||||
by_file_total_lines: dict[str, int] = {}
|
||||
by_file_max_end: dict[str, int] = {}
|
||||
by_file_tok_lines: dict[str, list[tuple[int, int]]] = defaultdict(list)
|
||||
|
||||
for _session, rc in reads:
|
||||
if rc.file_total:
|
||||
prev = by_file_total_lines.get(rc.base, 0)
|
||||
if rc.file_total > prev:
|
||||
by_file_total_lines[rc.base] = rc.file_total
|
||||
# Track max line ever observed.
|
||||
cands = [v for v in (rc.actual_b, rc.intent.end) if v is not None]
|
||||
if cands:
|
||||
cur = max(cands)
|
||||
prev = by_file_max_end.get(rc.base, 0)
|
||||
if cur > prev:
|
||||
by_file_max_end[rc.base] = cur
|
||||
# tok/line: only when we know how many lines came back AND tokens > 0.
|
||||
if rc.actual_a is not None and rc.actual_b is not None and rc.tokens > 0:
|
||||
n = rc.actual_b - rc.actual_a + 1
|
||||
if n > 0:
|
||||
by_file_tok_lines[rc.base].append((rc.tokens, n))
|
||||
|
||||
out: dict[str, FileStats] = {}
|
||||
files = (
|
||||
set(by_file_total_lines)
|
||||
| set(by_file_max_end)
|
||||
| set(by_file_tok_lines)
|
||||
)
|
||||
for f in files:
|
||||
size = by_file_total_lines.get(f) or by_file_max_end.get(f, 1)
|
||||
tok_lines = by_file_tok_lines.get(f, [])
|
||||
if tok_lines:
|
||||
tot_tok = sum(t for t, _ in tok_lines)
|
||||
tot_ln = sum(n for _, n in tok_lines)
|
||||
tpl = tot_tok / max(tot_ln, 1)
|
||||
else:
|
||||
tpl = FALLBACK_TPL
|
||||
out[f] = FileStats(size_lines=size, tokens_per_line=tpl,
|
||||
bytes_per_line=max(8.0, tpl * 4.0))
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Per-pair grouping
|
||||
|
||||
def group_pairs(reads: list[tuple[str, ReadCall]]) -> dict[tuple[str, str], list[ReadCall]]:
|
||||
by_pair: dict[tuple[str, str], list[ReadCall]] = defaultdict(list)
|
||||
for session, rc in reads:
|
||||
by_pair[(session, rc.base)].append(rc)
|
||||
# Already sorted by session, seq from the SQL.
|
||||
return by_pair
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Simulator
|
||||
|
||||
class Config(NamedTuple):
|
||||
default_page: int # lines returned for a bare path read
|
||||
line_cap: int # absolute max lines per read
|
||||
byte_cap: int # max bytes per read (modelled as line cap via bytes_per_line)
|
||||
summarize_min: int # min file size (lines) for summarizer to fire on bare reads
|
||||
# (-1 disables summarizer; 0 = always)
|
||||
|
||||
|
||||
def effective_returned(rc: ReadCall, fs: FileStats, cfg: Config) -> tuple[int, int] | None:
|
||||
"""Range the tool actually returns for one call under cfg.
|
||||
|
||||
Honours intent (what the agent asked for), then applies (default_page,
|
||||
line_cap, byte_cap, file_size) as bounding constraints. Returns None for
|
||||
intents without line bounds (conflicts/other).
|
||||
|
||||
The model is one observed call → one simulated call: we do NOT generate
|
||||
synthetic follow-ups when the cap shrinks the response. If the original
|
||||
session needed more lines, those follow-ups will appear as their own
|
||||
observed calls in the same pair.
|
||||
"""
|
||||
intent = rc.intent
|
||||
size = max(fs.size_lines, 1)
|
||||
if intent.kind == "bare":
|
||||
start, end_intent = 1, cfg.default_page
|
||||
elif intent.kind == "range":
|
||||
start = intent.start or 1
|
||||
end_intent = intent.end if intent.end is not None else (start + cfg.default_page - 1)
|
||||
elif intent.kind == "raw":
|
||||
start, end_intent = 1, size
|
||||
else:
|
||||
return None
|
||||
end = min(end_intent, size, start + cfg.line_cap - 1)
|
||||
if fs.bytes_per_line > 0:
|
||||
end = min(end, start + max(1, int(cfg.byte_cap / fs.bytes_per_line)) - 1)
|
||||
if end < start:
|
||||
end = start
|
||||
return (start, end)
|
||||
|
||||
|
||||
def cost_of_chunk(start: int, end: int, fs: FileStats, intent_kind: str, cfg: Config) -> float:
|
||||
"""Estimated result tokens for returning [start, end] of this file."""
|
||||
span = max(end - start + 1, 0)
|
||||
raw = span * fs.tokens_per_line
|
||||
if intent_kind == "bare" and cfg.summarize_min >= 0 and fs.size_lines >= cfg.summarize_min:
|
||||
# Calibrated from observed post-deploy summary-eligible reads:
|
||||
# tokens/line collapses to ~0.35× the verbatim rate.
|
||||
return raw * 0.35
|
||||
return raw
|
||||
|
||||
|
||||
def chunk_range(s: int, e: int, fs: FileStats, cfg: Config) -> list[tuple[int, int]]:
|
||||
"""Break [s, e] into chunks no larger than (line_cap, byte_cap)."""
|
||||
max_per_call = cfg.line_cap
|
||||
if fs.bytes_per_line > 0:
|
||||
max_per_call = min(max_per_call, max(1, int(cfg.byte_cap / fs.bytes_per_line)))
|
||||
out: list[tuple[int, int]] = []
|
||||
cur = s
|
||||
while cur <= e:
|
||||
end = min(cur + max_per_call - 1, e)
|
||||
out.append((cur, end))
|
||||
cur = end + 1
|
||||
return out
|
||||
|
||||
|
||||
def replay_pair(reads: list[ReadCall], fs: FileStats, cfg: Config) -> tuple[float, int]:
|
||||
"""Estimated tokens + calls for one (session, file).
|
||||
|
||||
Two-phase replay:
|
||||
1. For every observed call, compute the new returned range under cfg.
|
||||
Drop if fully covered; otherwise charge tokens + roundtrip and fold
|
||||
it into the simulated coverage.
|
||||
2. Compare simulated coverage against what the agent ACTUALLY received
|
||||
(union of observed returned ranges). Any shortfall is filled by
|
||||
synthetic chunks at (line_cap, byte_cap) granularity, charged at
|
||||
tokens + roundtrip. Phase 2 prevents the simulator from claiming
|
||||
free savings by silently returning fewer lines than the agent
|
||||
demonstrably needed.
|
||||
"""
|
||||
covered: list[tuple[int, int]] = []
|
||||
observed_needed: list[tuple[int, int]] = []
|
||||
total = 0.0
|
||||
kept = 0
|
||||
for rc in reads:
|
||||
if rc.actual_a is not None and rc.actual_b is not None and rc.actual_b >= rc.actual_a:
|
||||
observed_needed.append((rc.actual_a, rc.actual_b))
|
||||
ret = effective_returned(rc, fs, cfg)
|
||||
if ret is None:
|
||||
continue
|
||||
s, e = ret
|
||||
if contained(s, e, covered):
|
||||
continue
|
||||
total += cost_of_chunk(s, e, fs, rc.intent.kind, cfg)
|
||||
total += ROUNDTRIP_OVERHEAD
|
||||
kept += 1
|
||||
covered.append((s, e))
|
||||
covered = merge(covered)
|
||||
# Shortfall: lines the agent originally read that sim never delivered.
|
||||
observed_needed = merge(observed_needed)
|
||||
gaps: list[tuple[int, int]] = []
|
||||
for need_s, need_e in observed_needed:
|
||||
gaps.extend(subtract(need_s, need_e, covered))
|
||||
for gap_s, gap_e in gaps:
|
||||
for cs, ce in chunk_range(gap_s, gap_e, fs, cfg):
|
||||
total += cost_of_chunk(cs, ce, fs, "range", cfg)
|
||||
total += ROUNDTRIP_OVERHEAD
|
||||
kept += 1
|
||||
covered.append((cs, ce))
|
||||
covered = merge(covered)
|
||||
return total, kept
|
||||
|
||||
|
||||
def simulate(by_pair: dict, files: dict[str, FileStats], cfg: Config) -> tuple[float, int]:
|
||||
grand = 0.0
|
||||
kept = 0
|
||||
for (_session, base), reads in by_pair.items():
|
||||
fs = files.get(base) or FileStats(size_lines=1, tokens_per_line=FALLBACK_TPL,
|
||||
bytes_per_line=FALLBACK_BPL)
|
||||
t, k = replay_pair(reads, fs, cfg)
|
||||
grand += t
|
||||
kept += k
|
||||
return grand, kept
|
||||
|
||||
|
||||
def baseline_observed(reads: list[tuple[str, ReadCall]]) -> tuple[int, int]:
|
||||
"""Actual observed token spend (sum of result_tokens) and call count."""
|
||||
tot = sum(rc.tokens for _, rc in reads)
|
||||
return tot, len(reads)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Sweep + report
|
||||
|
||||
def sweep(by_pair: dict, files: dict[str, FileStats]) -> dict:
|
||||
defaults = [200, 300, 400, 500, 700, 1000, 1500, 2000, 3000]
|
||||
line_caps = [500, 1000, 1500, 2000, 3000, 5000]
|
||||
summary_thresholds = [-1, 0, 50, 150, 300, 600] # min file size to summarize
|
||||
|
||||
grid_tokens = np.zeros((len(defaults), len(line_caps)))
|
||||
grid_calls = np.zeros((len(defaults), len(line_caps)), dtype=np.int64)
|
||||
for i, D in enumerate(defaults):
|
||||
for j, L in enumerate(line_caps):
|
||||
cfg = Config(default_page=D, line_cap=L,
|
||||
byte_cap=CURRENT_BYTE_CAP, summarize_min=0)
|
||||
t, k = simulate(by_pair, files, cfg)
|
||||
grid_tokens[i, j] = t
|
||||
grid_calls[i, j] = k
|
||||
|
||||
# Best (D, L) for fixed summarize_min=0.
|
||||
flat = np.argmin(grid_tokens)
|
||||
i_best, j_best = np.unravel_index(flat, grid_tokens.shape)
|
||||
best_DL = (defaults[i_best], line_caps[j_best])
|
||||
|
||||
# Sweep summarize_min at best (D, L).
|
||||
sm_tokens = []
|
||||
for sm in summary_thresholds:
|
||||
cfg = Config(default_page=best_DL[0], line_cap=best_DL[1],
|
||||
byte_cap=CURRENT_BYTE_CAP, summarize_min=sm)
|
||||
t, k = simulate(by_pair, files, cfg)
|
||||
sm_tokens.append((sm, t, k))
|
||||
best_sm = min(sm_tokens, key=lambda x: x[1])
|
||||
|
||||
# Sweep byte_cap at best (D, L, summarize_min).
|
||||
byte_caps = [16 * 1024, 32 * 1024, 50 * 1024, 75 * 1024, 100 * 1024,
|
||||
150 * 1024, 200 * 1024]
|
||||
bc_tokens = []
|
||||
for bc in byte_caps:
|
||||
cfg = Config(default_page=best_DL[0], line_cap=best_DL[1],
|
||||
byte_cap=bc, summarize_min=best_sm[0])
|
||||
t, k = simulate(by_pair, files, cfg)
|
||||
bc_tokens.append((bc, t, k))
|
||||
best_bc = min(bc_tokens, key=lambda x: x[1])
|
||||
|
||||
# Final combined config (D, L, summarize_min, byte_cap) — should be the
|
||||
# global minimum given the order of dimensions.
|
||||
final_cfg = Config(default_page=best_DL[0], line_cap=best_DL[1],
|
||||
byte_cap=best_bc[0], summarize_min=best_sm[0])
|
||||
final_tokens, final_calls = simulate(by_pair, files, final_cfg)
|
||||
|
||||
return {
|
||||
"defaults": defaults,
|
||||
"line_caps": line_caps,
|
||||
"grid_tokens": grid_tokens,
|
||||
"grid_calls": grid_calls,
|
||||
"best_DL": best_DL,
|
||||
"summary_sweep": sm_tokens,
|
||||
"best_summary": best_sm,
|
||||
"byte_cap_sweep": bc_tokens,
|
||||
"best_byte_cap": best_bc,
|
||||
"final_cfg": final_cfg,
|
||||
"final_tokens": final_tokens,
|
||||
"final_calls": final_calls,
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Plotting
|
||||
|
||||
def plot(result: dict, baseline_sim: float, observed: int, out_path: Path) -> None:
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
plt.rcParams.update({"figure.dpi": 110, "font.size": 10})
|
||||
|
||||
fig, axes = plt.subplots(2, 2, figsize=(15, 11))
|
||||
|
||||
# Heatmap: relative to baseline (current config).
|
||||
ax = axes[0, 0]
|
||||
grid = result["grid_tokens"]
|
||||
rel = grid / baseline_sim
|
||||
im = ax.imshow(rel, cmap="RdYlGn_r", aspect="auto", origin="lower",
|
||||
vmin=max(0.6, rel.min()), vmax=min(1.4, rel.max() + 0.02))
|
||||
ax.set_xticks(range(len(result["line_caps"])))
|
||||
ax.set_xticklabels(result["line_caps"])
|
||||
ax.set_yticks(range(len(result["defaults"])))
|
||||
ax.set_yticklabels(result["defaults"])
|
||||
ax.set_xlabel("line cap (L)")
|
||||
ax.set_ylabel("default page (D)")
|
||||
ax.set_title("simulated read tokens / baseline\n(green = cheaper, red = more)")
|
||||
for i in range(grid.shape[0]):
|
||||
for j in range(grid.shape[1]):
|
||||
ax.text(j, i, f"{rel[i,j]:.2f}", ha="center", va="center",
|
||||
color="black", fontsize=8)
|
||||
fig.colorbar(im, ax=ax, fraction=0.05)
|
||||
# Highlight current and best.
|
||||
cur_i = result["defaults"].index(CURRENT_DEFAULT) if CURRENT_DEFAULT in result["defaults"] else None
|
||||
cur_j = result["line_caps"].index(CURRENT_LINE_CAP) if CURRENT_LINE_CAP in result["line_caps"] else None
|
||||
if cur_i is not None and cur_j is not None:
|
||||
ax.add_patch(mpatches.Rectangle((cur_j - 0.5, cur_i - 0.5), 1, 1,
|
||||
fill=False, edgecolor="#1d4ed8",
|
||||
linewidth=2.4, label="current"))
|
||||
best_i = result["defaults"].index(result["best_DL"][0])
|
||||
best_j = result["line_caps"].index(result["best_DL"][1])
|
||||
ax.add_patch(mpatches.Rectangle((best_j - 0.5, best_i - 0.5), 1, 1,
|
||||
fill=False, edgecolor="#000",
|
||||
linewidth=2.4, linestyle="--", label="optimum"))
|
||||
ax.legend(loc="upper right", frameon=True, fontsize=9)
|
||||
|
||||
# Default-page line (at best line cap).
|
||||
ax = axes[0, 1]
|
||||
best_L = result["best_DL"][1]
|
||||
j = result["line_caps"].index(best_L)
|
||||
col = grid[:, j] / baseline_sim
|
||||
ax.plot(result["defaults"], col, marker="o", linewidth=1.8, color="#0f766e")
|
||||
ax.axhline(1.0, color="#9ca3af", linestyle="--", linewidth=1.0)
|
||||
ax.axvline(CURRENT_DEFAULT, color="#1d4ed8", linestyle=":", linewidth=1.2, label="current default")
|
||||
best_D = result["best_DL"][0]
|
||||
ax.axvline(best_D, color="#000", linestyle="--", linewidth=1.4, label=f"optimum D={best_D}")
|
||||
ax.set_xscale("log")
|
||||
ax.set_xlabel("default page (D) — log scale")
|
||||
ax.set_ylabel("simulated tokens / baseline")
|
||||
ax.set_title(f"sensitivity to default page (line cap fixed at L={best_L})")
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
ax.legend(loc="best", frameon=False)
|
||||
|
||||
# Summarizer threshold sweep.
|
||||
ax = axes[1, 0]
|
||||
sm_data = result["summary_sweep"]
|
||||
xs = [str("off") if sm == -1 else ("always" if sm == 0 else f"≥{sm}") for sm, _, _ in sm_data]
|
||||
ys = [t / baseline_sim for _, t, _ in sm_data]
|
||||
bars = ax.bar(xs, ys, color=["#dc2626" if y > 1 else "#0f766e" for y in ys],
|
||||
edgecolor="#111", linewidth=0.5)
|
||||
for bar, y in zip(bars, ys):
|
||||
ax.text(bar.get_x() + bar.get_width() / 2, y + 0.005,
|
||||
f"{(y - 1) * 100:+.1f}%", ha="center", va="bottom", fontsize=9)
|
||||
ax.axhline(1.0, color="#9ca3af", linestyle="--", linewidth=1.0)
|
||||
ax.set_ylabel("simulated tokens / baseline")
|
||||
ax.set_xlabel("summarize files ≥ N lines")
|
||||
ax.set_title(f"summarizer threshold sweep (D={result['best_DL'][0]}, L={result['best_DL'][1]})")
|
||||
ax.grid(True, axis="y", alpha=0.25, linestyle="--")
|
||||
|
||||
# Byte cap sweep.
|
||||
ax = axes[1, 1]
|
||||
bc_data = result["byte_cap_sweep"]
|
||||
xs_kb = [bc // 1024 for bc, _, _ in bc_data]
|
||||
ys = [t / baseline_sim for _, t, _ in bc_data]
|
||||
ax.plot(xs_kb, ys, marker="o", linewidth=1.8, color="#7c3aed")
|
||||
ax.axhline(1.0, color="#9ca3af", linestyle="--", linewidth=1.0)
|
||||
ax.axvline(CURRENT_BYTE_CAP / 1024, color="#1d4ed8", linestyle=":",
|
||||
linewidth=1.2, label="current byte cap")
|
||||
best_bc_kb = result["best_byte_cap"][0] // 1024
|
||||
ax.axvline(best_bc_kb, color="#000", linestyle="--", linewidth=1.4,
|
||||
label=f"optimum {best_bc_kb} KB")
|
||||
ax.set_xlabel("byte cap (KB)")
|
||||
ax.set_ylabel("simulated tokens / baseline")
|
||||
ax.set_title("sensitivity to byte cap")
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
ax.legend(loc="best", frameon=False)
|
||||
|
||||
fig.suptitle(
|
||||
f"read tool config sweep — observed read spend {observed:,}, "
|
||||
f"simulator baseline {baseline_sim:,.0f}",
|
||||
fontsize=12, y=1.02,
|
||||
)
|
||||
fig.tight_layout()
|
||||
fig.savefig(out_path, bbox_inches="tight")
|
||||
plt.close(fig)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Report
|
||||
|
||||
def fmt_pct(x: float) -> str:
|
||||
if x >= 0:
|
||||
return f"+{x*100:.1f}%"
|
||||
return f"{x*100:.1f}%"
|
||||
|
||||
|
||||
def report(result: dict, baseline_sim: float, baseline_calls: int,
|
||||
observed: int, observed_calls: int) -> None:
|
||||
defaults = result["defaults"]
|
||||
line_caps = result["line_caps"]
|
||||
grid = result["grid_tokens"]
|
||||
calls = result["grid_calls"]
|
||||
|
||||
print(f"\nbaseline (current config: D={CURRENT_DEFAULT}, L={CURRENT_LINE_CAP}):")
|
||||
print(f" observed result tokens = {observed:>13,} (truth)")
|
||||
print(f" simulator under baseline = {baseline_sim:>13,.0f} "
|
||||
f"({fmt_pct((baseline_sim - observed) / observed)} vs observed)")
|
||||
print(f" observed read calls = {observed_calls:>13,}")
|
||||
print(f" simulator calls (baseline) = {baseline_calls:>11,}")
|
||||
|
||||
# Sweep table.
|
||||
print(f"\nsimulated read tokens (× of baseline) by (D, L):")
|
||||
header = " D \\ L " + " ".join(f"{L:>6}" for L in line_caps)
|
||||
print(header)
|
||||
for i, D in enumerate(defaults):
|
||||
row = " ".join(f"{grid[i,j]/baseline_sim:>6.2f}" for j in range(len(line_caps)))
|
||||
print(f" D={D:<6} {row}")
|
||||
print(f"\nbest (D, L) = {result['best_DL']} → "
|
||||
f"{grid[defaults.index(result['best_DL'][0]), line_caps.index(result['best_DL'][1])]:,.0f} tokens"
|
||||
f" ({fmt_pct(grid.min()/baseline_sim - 1)})")
|
||||
|
||||
# Summarizer threshold sweep at best (D, L).
|
||||
print(f"\nsummarizer threshold sweep at best (D, L) = {result['best_DL']}:")
|
||||
print(f" {'min_file_lines':<16} {'tokens':>12} {'vs baseline':>12}")
|
||||
for sm, t, k in result["summary_sweep"]:
|
||||
label = "off" if sm == -1 else ("always" if sm == 0 else f">={sm}")
|
||||
print(f" {label:<16} {t:>12,.0f} {fmt_pct(t/baseline_sim - 1):>12}")
|
||||
print(f"\nbest summarize_min = {result['best_summary'][0]} → "
|
||||
f"{result['best_summary'][1]:,.0f} tokens "
|
||||
f"({fmt_pct(result['best_summary'][1]/baseline_sim - 1)})")
|
||||
|
||||
# Byte cap sweep at (best D, L, summarize_min).
|
||||
print(f"\nbyte cap sweep at best (D, L, summarize_min):")
|
||||
print(f" {'byte_cap':<10} {'tokens':>12} {'vs baseline':>12}")
|
||||
for bc, t, k in result["byte_cap_sweep"]:
|
||||
print(f" {bc//1024:>4} KB {t:>12,.0f} {fmt_pct(t/baseline_sim - 1):>12}")
|
||||
print(f"\nbest byte_cap = {result['best_byte_cap'][0]//1024} KB → "
|
||||
f"{result['best_byte_cap'][1]:,.0f} tokens "
|
||||
f"({fmt_pct(result['best_byte_cap'][1]/baseline_sim - 1)})")
|
||||
|
||||
# Final recommendation.
|
||||
cfg = result["final_cfg"]
|
||||
print("\n" + "=" * 64)
|
||||
print(" RECOMMENDED CONFIG")
|
||||
print("=" * 64)
|
||||
print(f" read.defaultLimit {cfg.default_page} lines (current: {CURRENT_DEFAULT})")
|
||||
print(f" read.lineCap {cfg.line_cap} lines (current: {CURRENT_LINE_CAP})")
|
||||
print(f" read.byteCap {cfg.byte_cap//1024} KB (current: {CURRENT_BYTE_CAP//1024} KB)")
|
||||
sm_label = "off" if cfg.summarize_min == -1 else (
|
||||
"always" if cfg.summarize_min == 0 else f"only files ≥ {cfg.summarize_min} lines")
|
||||
print(f" read.summarizer {sm_label}")
|
||||
print(f" simulated savings {fmt_pct(result['final_tokens']/baseline_sim - 1)} "
|
||||
f"({baseline_sim - result['final_tokens']:,.0f} fewer tokens / window)")
|
||||
print(f" calls {result['final_calls']:,} "
|
||||
f"(baseline sim: {baseline_calls:,})")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Entry
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
|
||||
ap.add_argument("--since", default=DEFAULT_SINCE)
|
||||
args = ap.parse_args()
|
||||
|
||||
since = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
||||
since_ms = int(since.timestamp() * 1000)
|
||||
|
||||
if not DB_PATH.exists():
|
||||
sys.exit(f"db missing: {DB_PATH}")
|
||||
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
|
||||
print(f"loading reads since {args.since}...")
|
||||
reads = fetch_reads(conn, since_ms)
|
||||
conn.close()
|
||||
print(f" {len(reads):,} read calls")
|
||||
|
||||
# Per-file aggregates.
|
||||
files = aggregate_files(reads)
|
||||
sizes = np.array([f.size_lines for f in files.values()], dtype=np.int64)
|
||||
tpls = np.array([f.tokens_per_line for f in files.values()], dtype=float)
|
||||
print(f" {len(files):,} distinct files")
|
||||
print(f" file size p50={int(np.percentile(sizes,50))} "
|
||||
f"p90={int(np.percentile(sizes,90))} max={int(sizes.max())}")
|
||||
print(f" tokens/line p50={np.percentile(tpls,50):.2f} "
|
||||
f"p90={np.percentile(tpls,90):.2f} max={tpls.max():.2f}")
|
||||
|
||||
# Per-pair.
|
||||
by_pair = group_pairs(reads)
|
||||
print(f" {len(by_pair):,} (session, file) pairs")
|
||||
|
||||
# Baseline simulation.
|
||||
print("\nsimulating baseline...")
|
||||
baseline_cfg = Config(default_page=CURRENT_DEFAULT, line_cap=CURRENT_LINE_CAP,
|
||||
byte_cap=CURRENT_BYTE_CAP, summarize_min=0)
|
||||
baseline_sim, baseline_calls = simulate(by_pair, files, baseline_cfg)
|
||||
observed, observed_calls = baseline_observed(reads)
|
||||
|
||||
print("sweeping (default_page, line_cap, summarize_min)...")
|
||||
result = sweep(by_pair, files)
|
||||
|
||||
report(result, baseline_sim, baseline_calls, observed, observed_calls)
|
||||
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
out = OUT_DIR / "read-config-sweep.png"
|
||||
plot(result, baseline_sim, observed, out)
|
||||
print(f"\nwrote {out}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 68 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 136 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 139 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 150 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 110 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 92 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 182 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 204 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 171 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 158 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 107 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 160 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 600 KiB |
@@ -0,0 +1,356 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Did the read summarizer help?
|
||||
|
||||
The summarizer (commits 17ea26f86 / df8e41d7b / 7eaa9393311) shipped on
|
||||
2026-05-04. It only runs on reads with *no selector* (no `:N-M`, `:raw`,
|
||||
`:conflicts`, etc.). We classify each read call into:
|
||||
|
||||
summary-eligible : `path` has no selector after the final '/'
|
||||
selector : path ends in `:<digits>` / `:raw` / `:conflicts` / ...
|
||||
|
||||
Daily volume is normalized to *share of all tokens spent that day*, where
|
||||
the denominator = every tool's args + results + assistant text + assistant
|
||||
thinking + user messages. That removes the "I worked harder that day" effect.
|
||||
|
||||
Outputs to scripts/session-stats/out/read-summarizer-*.png.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import matplotlib.dates as mdates
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
|
||||
DB_PATH = Path.home() / ".omp" / "stats.db"
|
||||
OUT_DIR = Path(__file__).resolve().parent / "out"
|
||||
DAY_MS = 86_400_000
|
||||
|
||||
# Deploy boundary; override with --deploy YYYY-MM-DD.
|
||||
DEFAULT_DEPLOY = "2026-05-04"
|
||||
|
||||
COHORT_COLORS = {
|
||||
"summary-eligible": "#2563eb",
|
||||
"selector": "#9ca3af",
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Classification
|
||||
|
||||
def has_selector(path: str) -> bool:
|
||||
"""True iff `path` carries a read selector (`:50-200`, `:raw`, ...)."""
|
||||
if not path:
|
||||
return False
|
||||
tail = path.rsplit("/", 1)[-1]
|
||||
idx = tail.rfind(":")
|
||||
if idx < 0:
|
||||
return False
|
||||
suffix = tail[idx + 1 :]
|
||||
if not suffix:
|
||||
return False
|
||||
if suffix in ("raw", "conflicts"):
|
||||
return True
|
||||
return any(ch.isdigit() for ch in suffix)
|
||||
|
||||
|
||||
def cohort_of(arg_json: str | None) -> str | None:
|
||||
if not arg_json:
|
||||
return None
|
||||
try:
|
||||
obj = json.loads(arg_json)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
path = obj.get("path")
|
||||
if not isinstance(path, str):
|
||||
return None
|
||||
return "selector" if has_selector(path) else "summary-eligible"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Data
|
||||
|
||||
def fetch_read_calls(conn) -> dict[str, dict[str, np.ndarray]]:
|
||||
sql = """
|
||||
SELECT c.timestamp,
|
||||
c.arg_json,
|
||||
COALESCE(c.arg_tokens, 0) + COALESCE(r.result_tokens, 0) AS tok
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
WHERE c.tool_name = 'read'
|
||||
"""
|
||||
by: dict[str, list[tuple[int, int]]] = {k: [] for k in COHORT_COLORS}
|
||||
for ts, arg_json, tok in conn.execute(sql):
|
||||
c = cohort_of(arg_json)
|
||||
if c is None:
|
||||
continue
|
||||
by[c].append((ts, tok))
|
||||
out: dict[str, dict[str, np.ndarray]] = {}
|
||||
for c, rows in by.items():
|
||||
if not rows:
|
||||
out[c] = {"ts": np.array([], dtype=np.int64), "tok": np.array([], dtype=np.int64)}
|
||||
continue
|
||||
ts = np.fromiter((r[0] for r in rows), dtype=np.int64, count=len(rows))
|
||||
tok = np.fromiter((r[1] for r in rows), dtype=np.int64, count=len(rows))
|
||||
order = np.argsort(ts)
|
||||
out[c] = {"ts": ts[order], "tok": tok[order]}
|
||||
return out
|
||||
|
||||
|
||||
def fetch_daily_denominator(conn) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""Total tokens spent per day across every counted source."""
|
||||
sql = """
|
||||
SELECT bucket_ms, SUM(tok) FROM (
|
||||
SELECT CAST(c.timestamp / :day AS INTEGER) * :day AS bucket_ms,
|
||||
COALESCE(c.arg_tokens, 0) + COALESCE(r.result_tokens, 0) AS tok
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
UNION ALL
|
||||
SELECT CAST(timestamp / :day AS INTEGER) * :day,
|
||||
COALESCE(text_tokens,0) + COALESCE(thinking_tokens,0)
|
||||
FROM ss_assistant_msgs
|
||||
UNION ALL
|
||||
SELECT CAST(timestamp / :day AS INTEGER) * :day,
|
||||
COALESCE(text_tokens,0)
|
||||
FROM ss_user_msgs
|
||||
)
|
||||
GROUP BY bucket_ms
|
||||
ORDER BY bucket_ms
|
||||
"""
|
||||
rows = conn.execute(sql, {"day": DAY_MS}).fetchall()
|
||||
if not rows:
|
||||
return np.array([]), np.array([])
|
||||
bucket = np.fromiter((r[0] for r in rows), dtype=np.int64, count=len(rows))
|
||||
tot = np.fromiter((r[1] for r in rows), dtype=np.int64, count=len(rows))
|
||||
return bucket, tot
|
||||
|
||||
|
||||
def daily_sum(ts_ms: np.ndarray, tok: np.ndarray, day_axis: np.ndarray) -> np.ndarray:
|
||||
"""Sum `tok` per day (key = day_axis bucket_ms). Returns array len(day_axis)."""
|
||||
out = np.zeros(day_axis.size, dtype=np.int64)
|
||||
if ts_ms.size == 0:
|
||||
return out
|
||||
bucket = (ts_ms // DAY_MS) * DAY_MS
|
||||
idx = {int(d): i for i, d in enumerate(day_axis)}
|
||||
# Vectorize via searchsorted on a sorted day_axis (it is).
|
||||
pos = np.searchsorted(day_axis, bucket)
|
||||
for p, t, b in zip(pos, tok, bucket):
|
||||
if p < day_axis.size and day_axis[p] == b:
|
||||
out[p] += t
|
||||
return out
|
||||
|
||||
|
||||
def daily_percentile(ts_ms: np.ndarray, tok: np.ndarray, q: float) -> tuple[np.ndarray, np.ndarray]:
|
||||
if ts_ms.size == 0:
|
||||
return np.array([]), np.array([])
|
||||
day_idx = ts_ms // DAY_MS
|
||||
days = np.arange(day_idx.min(), day_idx.max() + 1)
|
||||
pct = np.full(days.size, np.nan)
|
||||
order = np.searchsorted(day_idx, days)
|
||||
order = np.append(order, ts_ms.size)
|
||||
for i in range(days.size):
|
||||
lo, hi = order[i], order[i + 1]
|
||||
if hi > lo:
|
||||
pct[i] = np.percentile(tok[lo:hi], q)
|
||||
dates = np.array([datetime.fromtimestamp(int(d) * DAY_MS / 1000, tz=timezone.utc) for d in days])
|
||||
return dates, pct
|
||||
|
||||
|
||||
def smooth_nan(y: np.ndarray, w: int) -> np.ndarray:
|
||||
if w <= 1 or y.size < w:
|
||||
return y
|
||||
mask = np.isfinite(y).astype(float)
|
||||
yf = np.where(mask > 0, y, 0.0)
|
||||
kernel = np.ones(w, dtype=float)
|
||||
num = np.convolve(yf, kernel, mode="same")
|
||||
den = np.convolve(mask, kernel, mode="same")
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
return np.where(den > 0, num / den, np.nan)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Plot helpers
|
||||
|
||||
def thousands(x: float, _p=0) -> str:
|
||||
if x >= 1000:
|
||||
return f"{x/1000:.1f}k"
|
||||
return f"{x:.0f}"
|
||||
|
||||
|
||||
def style_time(ax: plt.Axes, deploy: datetime) -> None:
|
||||
ax.xaxis.set_major_locator(mdates.MonthLocator())
|
||||
ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
ax.axvline(deploy, color="#dc2626", linestyle="--", linewidth=1.2, alpha=0.8)
|
||||
y1 = ax.get_ylim()[1] if ax.get_ylim()[1] > 0 else 1
|
||||
ax.text(deploy, y1, " summarizer\n deploy", color="#dc2626",
|
||||
va="top", ha="left", fontsize=9)
|
||||
|
||||
|
||||
def panel_share_stacked(ax: plt.Axes, reads, denom_dates, denom, deploy: datetime) -> None:
|
||||
"""Stacked area: per-day read-cohort share of total tokens."""
|
||||
series = []
|
||||
labels = []
|
||||
colors = []
|
||||
for cohort, color in COHORT_COLORS.items():
|
||||
d = reads[cohort]
|
||||
sums = daily_sum(d["ts"], d["tok"], denom_dates)
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
share = np.where(denom > 0, sums / denom, 0.0)
|
||||
series.append(smooth_nan(share, 7))
|
||||
labels.append(cohort)
|
||||
colors.append(color)
|
||||
x = np.array([datetime.fromtimestamp(int(d) / 1000, tz=timezone.utc) for d in denom_dates])
|
||||
ax.stackplot(x, series, labels=labels, colors=colors, alpha=0.85)
|
||||
ax.set_title("read share of daily token spend (7d MA)")
|
||||
ax.set_ylabel("share of all tokens that day")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda v, _: f"{v*100:.0f}%"))
|
||||
ax.set_ylim(0, None)
|
||||
ax.legend(loc="upper left", frameon=False)
|
||||
style_time(ax, deploy)
|
||||
|
||||
|
||||
def panel_share_line(ax: plt.Axes, reads, denom_dates, denom, deploy: datetime) -> None:
|
||||
"""Lines: each cohort's share, plus the combined total."""
|
||||
x = np.array([datetime.fromtimestamp(int(d) / 1000, tz=timezone.utc) for d in denom_dates])
|
||||
total = np.zeros(denom_dates.size, dtype=np.int64)
|
||||
for cohort, color in COHORT_COLORS.items():
|
||||
d = reads[cohort]
|
||||
sums = daily_sum(d["ts"], d["tok"], denom_dates)
|
||||
total += sums
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
share = np.where(denom > 0, sums / denom, 0.0)
|
||||
ax.plot(x, smooth_nan(share, 7), label=cohort, color=color, linewidth=1.7)
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
combined = np.where(denom > 0, total / denom, 0.0)
|
||||
ax.plot(x, smooth_nan(combined, 7), label="all reads", color="#111111", linewidth=2.2, linestyle="-")
|
||||
ax.set_title("read share by cohort (7d MA)")
|
||||
ax.set_ylabel("share of daily tokens")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda v, _: f"{v*100:.0f}%"))
|
||||
ax.set_ylim(0, None)
|
||||
ax.legend(loc="upper left", frameon=False)
|
||||
style_time(ax, deploy)
|
||||
|
||||
|
||||
def panel_per_call(ax: plt.Axes, reads, deploy: datetime, q: float, label_q: str) -> None:
|
||||
for cohort, color in COHORT_COLORS.items():
|
||||
d = reads[cohort]
|
||||
if d["ts"].size == 0:
|
||||
continue
|
||||
dates, pct = daily_percentile(d["ts"], d["tok"], q)
|
||||
ax.plot(dates, smooth_nan(pct, 7), label=f"{cohort} ({label_q})", color=color, linewidth=1.9)
|
||||
ax.set_title(f"daily {label_q} tokens per read call (7d MA)")
|
||||
ax.set_ylabel("tokens / call")
|
||||
ax.set_yscale("log")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(thousands))
|
||||
ax.legend(loc="upper left", frameon=False)
|
||||
style_time(ax, deploy)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Stats
|
||||
|
||||
def share_stats(reads, denom_dates, denom, deploy_ms: int) -> None:
|
||||
pre_mask = denom_dates < deploy_ms
|
||||
post_mask = denom_dates >= deploy_ms
|
||||
pre_total = int(denom[pre_mask].sum())
|
||||
post_total = int(denom[post_mask].sum())
|
||||
print(f"\nshare-of-day (pre vs post deploy):")
|
||||
print(f" denominator pre = {pre_total:>14,} tokens across {int(pre_mask.sum())} days")
|
||||
print(f" denominator post = {post_total:>14,} tokens across {int(post_mask.sum())} days")
|
||||
print(f" {'cohort':<22} {'pre share':>10} {'post share':>11} {'delta':>10}")
|
||||
grand_pre = 0
|
||||
grand_post = 0
|
||||
for cohort in COHORT_COLORS:
|
||||
d = reads[cohort]
|
||||
sums = daily_sum(d["ts"], d["tok"], denom_dates)
|
||||
pre = int(sums[pre_mask].sum())
|
||||
post = int(sums[post_mask].sum())
|
||||
grand_pre += pre
|
||||
grand_post += post
|
||||
pre_share = pre / pre_total if pre_total else 0
|
||||
post_share = post / post_total if post_total else 0
|
||||
print(f" {cohort:<22} {pre_share*100:>9.2f}% {post_share*100:>10.2f}% "
|
||||
f"{(post_share-pre_share)*100:>+9.2f}pp")
|
||||
pre_share = grand_pre / pre_total if pre_total else 0
|
||||
post_share = grand_post / post_total if post_total else 0
|
||||
print(f" {'all reads':<22} {pre_share*100:>9.2f}% {post_share*100:>10.2f}% "
|
||||
f"{(post_share-pre_share)*100:>+9.2f}pp")
|
||||
|
||||
|
||||
def per_call_stats(reads, deploy_ms: int) -> None:
|
||||
print(f"\nper-call stats (pre vs post deploy):")
|
||||
print(f" {'cohort':<22} {'window':<6} {'n':>9} {'p50':>7} {'p90':>7} {'mean':>8}")
|
||||
for cohort in COHORT_COLORS:
|
||||
d = reads[cohort]
|
||||
if d["ts"].size == 0:
|
||||
continue
|
||||
pre = d["tok"][d["ts"] < deploy_ms]
|
||||
post = d["tok"][d["ts"] >= deploy_ms]
|
||||
for name, arr in (("pre", pre), ("post", post)):
|
||||
if arr.size == 0:
|
||||
continue
|
||||
print(f" {cohort:<22} {name:<6} {arr.size:>9,} "
|
||||
f"{int(np.percentile(arr,50)):>7,} "
|
||||
f"{int(np.percentile(arr,90)):>7,} "
|
||||
f"{int(arr.mean()):>8,}")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Entry
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="read summarizer impact analysis")
|
||||
ap.add_argument("--deploy", default=DEFAULT_DEPLOY,
|
||||
help=f"deploy date YYYY-MM-DD (default {DEFAULT_DEPLOY})")
|
||||
args = ap.parse_args()
|
||||
|
||||
deploy = datetime.strptime(args.deploy, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
||||
deploy_ms = int(deploy.timestamp() * 1000)
|
||||
|
||||
if not DB_PATH.exists():
|
||||
sys.exit(f"db missing: {DB_PATH}")
|
||||
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
|
||||
reads = fetch_read_calls(conn)
|
||||
denom_dates, denom = fetch_daily_denominator(conn)
|
||||
conn.close()
|
||||
if denom_dates.size == 0:
|
||||
sys.exit("no daily totals available")
|
||||
|
||||
for c in COHORT_COLORS:
|
||||
print(f"{c:<22} calls={reads[c]['ts'].size:,}")
|
||||
|
||||
per_call_stats(reads, deploy_ms)
|
||||
share_stats(reads, denom_dates, denom, deploy_ms)
|
||||
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
plt.rcParams.update({"figure.dpi": 110, "font.size": 10})
|
||||
|
||||
fig, axes = plt.subplots(2, 2, figsize=(15, 9))
|
||||
panel_share_stacked(axes[0, 0], reads, denom_dates, denom, deploy)
|
||||
panel_share_line(axes[0, 1], reads, denom_dates, denom, deploy)
|
||||
panel_per_call(axes[1, 0], reads, deploy, q=50, label_q="p50")
|
||||
panel_per_call(axes[1, 1], reads, deploy, q=90, label_q="p90")
|
||||
fig.suptitle(f"read summarizer impact — deploy = {args.deploy}", fontsize=13, y=0.995)
|
||||
fig.tight_layout()
|
||||
p = OUT_DIR / "read-summarizer.png"
|
||||
fig.savefig(p, bbox_inches="tight")
|
||||
plt.close(fig)
|
||||
print(f"\nwrote {p}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,408 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Plot token-usage trends for the top N tools from ~/.omp/stats.db.
|
||||
|
||||
Reads ss_tool_calls + ss_tool_results and renders:
|
||||
1. daily total tokens (args + results) -- stacked area
|
||||
2. daily call count -- lines
|
||||
3. daily mean tokens per call -- lines (log y)
|
||||
4. cumulative tokens -- lines
|
||||
5. weekly median tokens-per-call -- lines (log y)
|
||||
6. per-call token histogram (overall) -- log-log step
|
||||
|
||||
`grep` is folded into `search` (old → new name). Tools are picked as the top
|
||||
N by total tokens (default 10); override with --top N or --tools a,b,c.
|
||||
|
||||
Output: scripts/session-stats/out/tool-trends.png + standalone panels.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sqlite3
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
import matplotlib.dates as mdates
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
|
||||
DB_PATH = Path.home() / ".omp" / "stats.db"
|
||||
OUT_DIR = Path(__file__).resolve().parent / "out"
|
||||
|
||||
DAY_MS = 86_400_000
|
||||
WEEK_MS = 7 * DAY_MS
|
||||
|
||||
# tool_name normalization: old names → canonical names.
|
||||
TOOL_ALIAS = {"grep": "search"}
|
||||
|
||||
# 10-class qualitative palette (tab10) — distinct hues for line + area work.
|
||||
PALETTE = [
|
||||
"#1f77b4", "#d62728", "#2ca02c", "#ff7f0e", "#9467bd",
|
||||
"#8c564b", "#17becf", "#e377c2", "#bcbd22", "#7f7f7f",
|
||||
"#393b79", "#637939",
|
||||
]
|
||||
|
||||
|
||||
def normalize_case_sql(col: str) -> str:
|
||||
"""Build a CASE expression that maps aliases to canonical names."""
|
||||
if not TOOL_ALIAS:
|
||||
return col
|
||||
whens = " ".join(f"WHEN '{k}' THEN '{v}'" for k, v in TOOL_ALIAS.items())
|
||||
return f"CASE {col} {whens} ELSE {col} END"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Data access
|
||||
|
||||
def _connect() -> sqlite3.Connection:
|
||||
if not DB_PATH.exists():
|
||||
sys.exit(f"db missing: {DB_PATH}")
|
||||
return sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
|
||||
|
||||
|
||||
def pick_top_tools(conn: sqlite3.Connection, top: int) -> list[str]:
|
||||
norm = normalize_case_sql("c.tool_name")
|
||||
sql = f"""
|
||||
SELECT {norm} AS tool,
|
||||
SUM(COALESCE(c.arg_tokens,0) + COALESCE(r.result_tokens,0)) AS total
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
GROUP BY tool
|
||||
ORDER BY total DESC
|
||||
LIMIT ?
|
||||
"""
|
||||
return [row[0] for row in conn.execute(sql, (top,))]
|
||||
|
||||
|
||||
def fetch_daily(conn: sqlite3.Connection, tools: list[str]) -> dict:
|
||||
placeholders = ",".join("?" * len(tools))
|
||||
norm = normalize_case_sql("c.tool_name")
|
||||
sql = f"""
|
||||
SELECT {norm} AS tool,
|
||||
CAST(c.timestamp / ? AS INTEGER) * ? AS bucket_ms,
|
||||
COUNT(*) AS calls,
|
||||
COALESCE(SUM(c.arg_tokens), 0) AS arg_tokens,
|
||||
COALESCE(SUM(r.result_tokens), 0) AS result_tokens
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
WHERE {norm} IN ({placeholders})
|
||||
GROUP BY tool, bucket_ms
|
||||
ORDER BY bucket_ms
|
||||
"""
|
||||
rows = conn.execute(sql, (DAY_MS, DAY_MS, *tools)).fetchall()
|
||||
if not rows:
|
||||
sys.exit(f"no rows for tools={tools}")
|
||||
|
||||
all_days = sorted({r[1] for r in rows})
|
||||
start, end = all_days[0], all_days[-1]
|
||||
day_axis = list(range(start, end + DAY_MS, DAY_MS))
|
||||
idx = {d: i for i, d in enumerate(day_axis)}
|
||||
|
||||
n = len(day_axis)
|
||||
per_tool = {
|
||||
t: {
|
||||
"calls": np.zeros(n, dtype=np.int64),
|
||||
"args": np.zeros(n, dtype=np.int64),
|
||||
"results": np.zeros(n, dtype=np.int64),
|
||||
}
|
||||
for t in tools
|
||||
}
|
||||
for tool, bucket_ms, calls, args, results in rows:
|
||||
i = idx[bucket_ms]
|
||||
per_tool[tool]["calls"][i] = calls
|
||||
per_tool[tool]["args"][i] = args
|
||||
per_tool[tool]["results"][i] = results
|
||||
|
||||
dates = np.array(
|
||||
[datetime.fromtimestamp(d / 1000, tz=timezone.utc) for d in day_axis]
|
||||
)
|
||||
return {"dates": dates, **per_tool}
|
||||
|
||||
|
||||
def fetch_per_call(conn: sqlite3.Connection, tools: list[str]) -> dict[str, dict]:
|
||||
placeholders = ",".join("?" * len(tools))
|
||||
norm = normalize_case_sql("c.tool_name")
|
||||
sql = f"""
|
||||
SELECT {norm} AS tool,
|
||||
c.timestamp,
|
||||
COALESCE(c.arg_tokens, 0) + COALESCE(r.result_tokens, 0) AS total
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
WHERE {norm} IN ({placeholders})
|
||||
"""
|
||||
by_tool: dict[str, list[tuple[int, int]]] = {t: [] for t in tools}
|
||||
for tool, ts, total in conn.execute(sql, tuple(tools)):
|
||||
by_tool[tool].append((ts, total))
|
||||
|
||||
out: dict[str, dict] = {}
|
||||
for t, rows in by_tool.items():
|
||||
if not rows:
|
||||
out[t] = {"ts": np.array([], dtype=np.int64), "tok": np.array([], dtype=np.int64)}
|
||||
continue
|
||||
ts = np.fromiter((r[0] for r in rows), dtype=np.int64, count=len(rows))
|
||||
tok = np.fromiter((r[1] for r in rows), dtype=np.int64, count=len(rows))
|
||||
order = np.argsort(ts)
|
||||
out[t] = {"ts": ts[order], "tok": tok[order]}
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Helpers
|
||||
|
||||
def smooth(y: np.ndarray, w: int = 7) -> np.ndarray:
|
||||
if w <= 1 or len(y) < w:
|
||||
return y.astype(float)
|
||||
kernel = np.ones(w, dtype=float) / w
|
||||
return np.convolve(y.astype(float), kernel, mode="same")
|
||||
|
||||
|
||||
def smooth_nan(y: np.ndarray, w: int = 7) -> np.ndarray:
|
||||
if w <= 1 or len(y) < w:
|
||||
return y
|
||||
mask = np.isfinite(y).astype(float)
|
||||
yf = np.where(mask > 0, y, 0.0)
|
||||
kernel = np.ones(w, dtype=float)
|
||||
num = np.convolve(yf, kernel, mode="same")
|
||||
den = np.convolve(mask, kernel, mode="same")
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
return np.where(den > 0, num / den, np.nan)
|
||||
|
||||
|
||||
def millions(x: float, _pos: int = 0) -> str:
|
||||
if x >= 1e6:
|
||||
return f"{x / 1e6:.1f}M"
|
||||
if x >= 1e3:
|
||||
return f"{x / 1e3:.0f}k"
|
||||
return f"{x:.0f}"
|
||||
|
||||
|
||||
def style_time_axis(ax: plt.Axes) -> None:
|
||||
ax.xaxis.set_major_locator(mdates.MonthLocator())
|
||||
ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
|
||||
ax.tick_params(axis="x", rotation=0)
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
|
||||
|
||||
def weekly_median(ts_ms: np.ndarray, tok: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""Returns (week_dates, p50) — both 1-d, same length."""
|
||||
if ts_ms.size == 0:
|
||||
return np.array([]), np.array([])
|
||||
week_idx = ts_ms // WEEK_MS
|
||||
weeks = np.arange(week_idx.min(), week_idx.max() + 1)
|
||||
p50 = np.full(weeks.size, np.nan)
|
||||
order = np.searchsorted(week_idx, weeks)
|
||||
order = np.append(order, ts_ms.size)
|
||||
for i in range(weeks.size):
|
||||
lo, hi = order[i], order[i + 1]
|
||||
if hi > lo:
|
||||
p50[i] = np.percentile(tok[lo:hi], 50)
|
||||
week_dates = np.array(
|
||||
[datetime.fromtimestamp(int(w) * WEEK_MS / 1000, tz=timezone.utc) for w in weeks]
|
||||
)
|
||||
return week_dates, p50
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Panels
|
||||
|
||||
def panel_total_tokens(ax: plt.Axes, daily: dict, tools: list[str], colors: dict) -> None:
|
||||
dates = daily["dates"]
|
||||
series = [smooth(daily[t]["args"] + daily[t]["results"]) for t in tools]
|
||||
ax.stackplot(dates, series, labels=tools, colors=[colors[t] for t in tools], alpha=0.9)
|
||||
ax.set_title("Daily token volume (args + results, 7d MA)")
|
||||
ax.set_ylabel("tokens / day")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(millions))
|
||||
ax.legend(loc="upper left", frameon=False, ncol=2, fontsize=9)
|
||||
style_time_axis(ax)
|
||||
|
||||
|
||||
def panel_call_counts(ax: plt.Axes, daily: dict, tools: list[str], colors: dict) -> None:
|
||||
dates = daily["dates"]
|
||||
for t in tools:
|
||||
ax.plot(dates, smooth(daily[t]["calls"]), label=t, color=colors[t], linewidth=1.6)
|
||||
ax.set_title("Daily call count (7d MA)")
|
||||
ax.set_ylabel("calls / day")
|
||||
ax.legend(loc="upper left", frameon=False, ncol=2, fontsize=9)
|
||||
style_time_axis(ax)
|
||||
|
||||
|
||||
def panel_mean_per_call(ax: plt.Axes, daily: dict, tools: list[str], colors: dict) -> None:
|
||||
dates = daily["dates"]
|
||||
for t in tools:
|
||||
totals = daily[t]["args"] + daily[t]["results"]
|
||||
calls = daily[t]["calls"].astype(float)
|
||||
with np.errstate(divide="ignore", invalid="ignore"):
|
||||
mean = np.where(calls > 0, totals / calls, np.nan)
|
||||
ax.plot(dates, smooth_nan(mean), label=t, color=colors[t], linewidth=1.6)
|
||||
ax.set_title("Mean tokens per call (7d MA)")
|
||||
ax.set_ylabel("tokens / call")
|
||||
ax.set_yscale("log")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(millions))
|
||||
ax.legend(loc="upper left", frameon=False, ncol=2, fontsize=9)
|
||||
style_time_axis(ax)
|
||||
|
||||
|
||||
def panel_cumulative(ax: plt.Axes, daily: dict, tools: list[str], colors: dict) -> None:
|
||||
dates = daily["dates"]
|
||||
for t in tools:
|
||||
totals = daily[t]["args"] + daily[t]["results"]
|
||||
ax.plot(dates, np.cumsum(totals), label=t, color=colors[t], linewidth=1.6)
|
||||
ax.set_title("Cumulative tokens")
|
||||
ax.set_ylabel("tokens (total)")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(millions))
|
||||
ax.legend(loc="upper left", frameon=False, ncol=2, fontsize=9)
|
||||
style_time_axis(ax)
|
||||
|
||||
|
||||
def panel_weekly_median(ax: plt.Axes, per_call: dict, tools: list[str], colors: dict) -> None:
|
||||
for t in tools:
|
||||
w, p50 = weekly_median(per_call[t]["ts"], per_call[t]["tok"])
|
||||
if w.size == 0:
|
||||
continue
|
||||
ax.plot(w, p50, label=t, color=colors[t], linewidth=1.7)
|
||||
ax.set_title("Weekly median tokens / call")
|
||||
ax.set_ylabel("tokens / call (p50)")
|
||||
ax.set_yscale("log")
|
||||
ax.yaxis.set_major_formatter(plt.FuncFormatter(millions))
|
||||
ax.legend(loc="upper left", frameon=False, ncol=2, fontsize=9)
|
||||
style_time_axis(ax)
|
||||
|
||||
|
||||
def panel_histogram(ax: plt.Axes, per_call: dict, tools: list[str], colors: dict) -> None:
|
||||
all_tok = np.concatenate([per_call[t]["tok"] for t in tools if per_call[t]["tok"].size])
|
||||
if all_tok.size == 0:
|
||||
return
|
||||
hi = max(all_tok.max(), 10)
|
||||
bins = np.logspace(0, np.log10(hi), 60)
|
||||
for t in tools:
|
||||
tok = per_call[t]["tok"]
|
||||
if tok.size == 0:
|
||||
continue
|
||||
p50 = int(np.percentile(tok, 50))
|
||||
p99 = int(np.percentile(tok, 99))
|
||||
ax.hist(
|
||||
np.maximum(tok, 1),
|
||||
bins=bins,
|
||||
histtype="step",
|
||||
linewidth=1.5,
|
||||
color=colors[t],
|
||||
label=f"{t} (n={tok.size:,}, p50={p50}, p99={p99})",
|
||||
)
|
||||
ax.set_xscale("log")
|
||||
ax.set_yscale("log")
|
||||
ax.set_xlabel("tokens / call")
|
||||
ax.set_ylabel("calls")
|
||||
ax.xaxis.set_major_formatter(plt.FuncFormatter(millions))
|
||||
ax.set_title("Per-call token histogram (whole window)")
|
||||
ax.legend(loc="upper right", frameon=False, fontsize=8, ncol=1)
|
||||
ax.grid(True, which="both", alpha=0.2, linestyle="--")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Entry
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
|
||||
ap.add_argument("--top", type=int, default=10, help="top N tools by total tokens")
|
||||
ap.add_argument(
|
||||
"--tools",
|
||||
type=str,
|
||||
default=None,
|
||||
help="comma-separated tools to plot (overrides --top)",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
|
||||
conn = _connect()
|
||||
if args.tools:
|
||||
tools = [t.strip() for t in args.tools.split(",") if t.strip()]
|
||||
else:
|
||||
tools = pick_top_tools(conn, args.top)
|
||||
if not tools:
|
||||
sys.exit("no tools selected")
|
||||
if len(tools) > len(PALETTE):
|
||||
sys.exit(f"palette has {len(PALETTE)} colors but {len(tools)} tools requested")
|
||||
|
||||
colors = {t: PALETTE[i] for i, t in enumerate(tools)}
|
||||
print(f"plotting tools (ranked): {', '.join(tools)}")
|
||||
|
||||
daily = fetch_daily(conn, tools)
|
||||
per_call = fetch_per_call(conn, tools)
|
||||
conn.close()
|
||||
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
plt.rcParams.update({"figure.dpi": 110, "font.size": 10})
|
||||
|
||||
# Combined 3x2 dashboard.
|
||||
fig, axes = plt.subplots(3, 2, figsize=(15, 13))
|
||||
panel_total_tokens(axes[0, 0], daily, tools, colors)
|
||||
panel_call_counts(axes[0, 1], daily, tools, colors)
|
||||
panel_mean_per_call(axes[1, 0], daily, tools, colors)
|
||||
panel_cumulative(axes[1, 1], daily, tools, colors)
|
||||
panel_weekly_median(axes[2, 0], per_call, tools, colors)
|
||||
panel_histogram(axes[2, 1], per_call, tools, colors)
|
||||
fig.suptitle(
|
||||
f"top {len(tools)} tools — token-usage trends "
|
||||
f"({daily['dates'][0].date()} → {daily['dates'][-1].date()})",
|
||||
fontsize=13,
|
||||
y=0.995,
|
||||
)
|
||||
fig.tight_layout()
|
||||
combined = OUT_DIR / "tool-trends.png"
|
||||
fig.savefig(combined, bbox_inches="tight")
|
||||
plt.close(fig)
|
||||
print(f"wrote {combined}")
|
||||
|
||||
panels: tuple[tuple[str, Callable, dict], ...] = (
|
||||
("daily-tokens", panel_total_tokens, daily),
|
||||
("daily-calls", panel_call_counts, daily),
|
||||
("tokens-per-call", panel_mean_per_call, daily),
|
||||
("cumulative-tokens", panel_cumulative, daily),
|
||||
("per-call-median", panel_weekly_median, per_call),
|
||||
("per-call-histogram", panel_histogram, per_call),
|
||||
)
|
||||
for name, fn, src in panels:
|
||||
f2, ax = plt.subplots(figsize=(11, 5))
|
||||
fn(ax, src, tools, colors)
|
||||
f2.tight_layout()
|
||||
p = OUT_DIR / f"{name}.png"
|
||||
f2.savefig(p, bbox_inches="tight")
|
||||
plt.close(f2)
|
||||
print(f"wrote {p}")
|
||||
|
||||
# Summary.
|
||||
print()
|
||||
print("totals over the window:")
|
||||
header = f" {'tool':<14} {'calls':>9} {'total':>14} {'p50':>6} {'p90':>7} {'p99':>8} {'max':>9}"
|
||||
print(header)
|
||||
print(" " + "-" * (len(header) - 2))
|
||||
for t in tools:
|
||||
a = int(daily[t]["args"].sum())
|
||||
r = int(daily[t]["results"].sum())
|
||||
c = int(daily[t]["calls"].sum())
|
||||
tok = per_call[t]["tok"]
|
||||
if tok.size:
|
||||
p50 = int(np.percentile(tok, 50))
|
||||
p90 = int(np.percentile(tok, 90))
|
||||
p99 = int(np.percentile(tok, 99))
|
||||
mx = int(tok.max())
|
||||
else:
|
||||
p50 = p90 = p99 = mx = 0
|
||||
print(
|
||||
f" {t:<14} {c:>9,} {a + r:>14,} {p50:>6,} {p90:>7,} {p99:>8,} {mx:>9,}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,628 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Optimize read-tool line-window settings from historical session behaviour.
|
||||
|
||||
This is a counterfactual replay over post-summarizer read calls. For each
|
||||
(session, file), reads are replayed in order while maintaining a line-coverage
|
||||
map. A candidate config changes the interval delivered by each bounded/default
|
||||
read. If a later requested interval is already covered, that later read would
|
||||
have been avoided.
|
||||
|
||||
Modelled config dimensions:
|
||||
defaultLimit lines returned by bare reads and open selectors (`:N`)
|
||||
maxLines hard line cap for one read result
|
||||
leadingContext lines before explicit offsets/ranges
|
||||
trailingContext lines after explicit bounded ranges
|
||||
|
||||
The replay reports estimated token cost, read calls, avoided follow-ups,
|
||||
truncations, and a Pareto frontier.
|
||||
|
||||
Output:
|
||||
scripts/session-stats/out/read-optimizer.png
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
|
||||
DB_PATH = Path.home() / ".omp" / "stats.db"
|
||||
OUT_DIR = Path(__file__).resolve().parent / "out"
|
||||
DEFAULT_SINCE = "2026-05-04"
|
||||
|
||||
# Current code defaults, from packages/coding-agent/src/tools/read.ts and
|
||||
# packages/coding-agent/src/config/settings-schema.ts.
|
||||
CURRENT_DEFAULT = 500
|
||||
CURRENT_MAX_LINES = 3000
|
||||
CURRENT_LEADING = 3
|
||||
CURRENT_TRAILING = 3
|
||||
CURRENT_MAX_BYTES = 50 * 1024
|
||||
READ_MAX_COLUMN = 768
|
||||
|
||||
_RANGE_RE = re.compile(r"^(\d+)(?:([-+])(\d+))?$")
|
||||
TEXT_EXTS = {
|
||||
".ts", ".tsx", ".js", ".jsx", ".mts", ".cts", ".mjs", ".cjs",
|
||||
".rs", ".go", ".py", ".rb", ".java", ".kt", ".kts", ".c", ".cc",
|
||||
".cpp", ".h", ".hpp", ".cs", ".swift", ".php", ".lua", ".sh",
|
||||
".bash", ".zsh", ".fish", ".md", ".txt", ".json", ".jsonc", ".json5",
|
||||
".yaml", ".yml", ".toml", ".xml", ".html", ".css", ".scss", ".sql",
|
||||
".adoc", ".typ", ".rsx", ".vue", ".svelte", ".dockerfile", "",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReadCall:
|
||||
session: str
|
||||
file: str
|
||||
seq: int
|
||||
kind: str # explicit | open | default | raw | conflicts | other
|
||||
start: int | None
|
||||
end: int | None
|
||||
arg_tokens: int
|
||||
result_tokens: int
|
||||
current_lines: int
|
||||
token_per_line: float
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
default: int
|
||||
max_lines: int
|
||||
leading: int
|
||||
trailing: int
|
||||
|
||||
def label(self) -> str:
|
||||
return f"D{self.default}/M{self.max_lines}/L{self.leading}/T{self.trailing}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReplayResult:
|
||||
config: Config
|
||||
tokens: float
|
||||
calls: int
|
||||
skipped_calls: int
|
||||
truncations: int
|
||||
bytes_limited: int
|
||||
selector_tokens: float
|
||||
selector_calls: int
|
||||
selector_skipped: int
|
||||
selector_groups: int
|
||||
selector_groups_single_call: int
|
||||
selector_groups_all_covered_by_first: int
|
||||
default_tokens: float
|
||||
default_calls: int
|
||||
default_skipped: int
|
||||
raw_tokens: float
|
||||
raw_calls: int
|
||||
|
||||
@property
|
||||
def first_cover_rate(self) -> float:
|
||||
if self.selector_groups == 0:
|
||||
return 0.0
|
||||
return self.selector_groups_all_covered_by_first / self.selector_groups
|
||||
|
||||
|
||||
def parse_int_list(spec: str) -> list[int]:
|
||||
out: list[int] = []
|
||||
for part in spec.split(","):
|
||||
part = part.strip()
|
||||
if not part:
|
||||
continue
|
||||
out.append(int(part))
|
||||
return out
|
||||
|
||||
|
||||
def parse_path_selector(path: str) -> tuple[str, str, int | None, int | None]:
|
||||
if not path:
|
||||
return path, "other", None, None
|
||||
tail_idx = path.rfind("/")
|
||||
tail = path[tail_idx + 1 :]
|
||||
colon = tail.rfind(":")
|
||||
if colon < 0:
|
||||
return path, "default", 1, CURRENT_DEFAULT
|
||||
|
||||
suffix = tail[colon + 1 :]
|
||||
base = path[: tail_idx + 1] + tail[:colon] if tail_idx >= 0 else tail[:colon]
|
||||
if suffix == "raw":
|
||||
return base, "raw", None, None
|
||||
if suffix == "conflicts":
|
||||
return base, "conflicts", None, None
|
||||
|
||||
m = _RANGE_RE.match(suffix)
|
||||
if not m:
|
||||
return path, "default", 1, CURRENT_DEFAULT
|
||||
start = int(m.group(1))
|
||||
op = m.group(2)
|
||||
nval = m.group(3)
|
||||
if op == "-" and nval is not None:
|
||||
return base, "explicit", start, max(start, int(nval))
|
||||
if op == "+" and nval is not None:
|
||||
return base, "explicit", start, start + max(1, int(nval)) - 1
|
||||
return base, "open", start, start + CURRENT_DEFAULT - 1
|
||||
|
||||
|
||||
def current_line_count(kind: str, start: int | None, end: int | None) -> int:
|
||||
if kind not in ("explicit", "open", "default") or start is None or end is None:
|
||||
return 0
|
||||
if kind == "explicit":
|
||||
requested = max(1, end - start + 1)
|
||||
leading = min(start - 1, CURRENT_LEADING) if start > 1 else 0
|
||||
return min(requested + leading + CURRENT_TRAILING, CURRENT_MAX_LINES)
|
||||
if kind == "open":
|
||||
leading = min(start - 1, CURRENT_LEADING) if start > 1 else 0
|
||||
return min(CURRENT_DEFAULT + leading, CURRENT_MAX_LINES)
|
||||
return min(CURRENT_DEFAULT, CURRENT_MAX_LINES)
|
||||
|
||||
|
||||
def parse_call(row) -> ReadCall | None:
|
||||
session, seq, arg_json, arg_tokens, result_tokens = row
|
||||
try:
|
||||
obj = json.loads(arg_json or "{}")
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
path = obj.get("path")
|
||||
if not isinstance(path, str):
|
||||
return None
|
||||
base, kind, start, end = parse_path_selector(path)
|
||||
|
||||
# Legacy/bridge fields override a bare path.
|
||||
if kind == "default":
|
||||
offset = obj.get("offset")
|
||||
limit = obj.get("limit")
|
||||
if isinstance(offset, int) and offset >= 1 and isinstance(limit, int) and limit >= 1:
|
||||
kind = "explicit"
|
||||
start = offset
|
||||
end = offset + limit - 1
|
||||
elif isinstance(offset, int) and offset >= 1:
|
||||
kind = "open"
|
||||
start = offset
|
||||
end = offset + CURRENT_DEFAULT - 1
|
||||
|
||||
if not base or base.endswith("/") or "://" in base:
|
||||
return None
|
||||
ext = Path(base).suffix.lower()
|
||||
if ext not in TEXT_EXTS:
|
||||
# Keep unknown extension text if it has line selectors, skip obvious binary-ish paths.
|
||||
if kind not in ("explicit", "open", "default"):
|
||||
return None
|
||||
|
||||
current_lines = current_line_count(kind, start, end)
|
||||
rtok = int(result_tokens or 0)
|
||||
if current_lines > 0:
|
||||
# Include the observed framing/line-number overhead in a per-line rate.
|
||||
# Clamp avoids a one-line error response implying giant line cost.
|
||||
token_per_line = min(100.0, max(0.25, rtok / current_lines))
|
||||
else:
|
||||
token_per_line = 0.0
|
||||
return ReadCall(
|
||||
session=str(session),
|
||||
file=base,
|
||||
seq=int(seq),
|
||||
kind=kind,
|
||||
start=start,
|
||||
end=end,
|
||||
arg_tokens=int(arg_tokens or 0),
|
||||
result_tokens=rtok,
|
||||
current_lines=current_lines,
|
||||
token_per_line=token_per_line,
|
||||
)
|
||||
|
||||
|
||||
def requested_interval(call: ReadCall, cfg: Config) -> tuple[int, int] | None:
|
||||
if call.kind == "explicit" and call.start is not None and call.end is not None:
|
||||
return call.start, call.end
|
||||
if call.kind == "open" and call.start is not None:
|
||||
return call.start, call.start + cfg.default - 1
|
||||
if call.kind == "default":
|
||||
return 1, cfg.default
|
||||
return None
|
||||
|
||||
|
||||
def delivered_interval(call: ReadCall, cfg: Config) -> tuple[int, int] | None:
|
||||
req = requested_interval(call, cfg)
|
||||
if req is None:
|
||||
return None
|
||||
s, e = req
|
||||
if call.kind == "explicit":
|
||||
start = max(1, s - cfg.leading) if s > 1 else 1
|
||||
requested = max(1, e - s + 1)
|
||||
lines = min(requested + (s - start) + cfg.trailing, cfg.max_lines)
|
||||
return start, start + lines - 1
|
||||
if call.kind == "open":
|
||||
start = max(1, s - cfg.leading) if s > 1 else 1
|
||||
lines = min(cfg.default + (s - start), cfg.max_lines)
|
||||
return start, start + lines - 1
|
||||
if call.kind == "default":
|
||||
lines = min(cfg.default, cfg.max_lines)
|
||||
return 1, lines
|
||||
return None
|
||||
|
||||
|
||||
def is_covered(intervals: list[tuple[int, int]], target: tuple[int, int]) -> bool:
|
||||
s, e = target
|
||||
for a, b in intervals:
|
||||
if a <= s and e <= b:
|
||||
return True
|
||||
if a > s:
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
def add_interval(intervals: list[tuple[int, int]], item: tuple[int, int]) -> list[tuple[int, int]]:
|
||||
s, e = item
|
||||
out: list[tuple[int, int]] = []
|
||||
placed = False
|
||||
for a, b in intervals:
|
||||
if b + 1 < s:
|
||||
out.append((a, b))
|
||||
elif e + 1 < a:
|
||||
if not placed:
|
||||
out.append((s, e))
|
||||
placed = True
|
||||
out.append((a, b))
|
||||
else:
|
||||
s = min(s, a)
|
||||
e = max(e, b)
|
||||
if not placed:
|
||||
out.append((s, e))
|
||||
return out
|
||||
|
||||
|
||||
def estimate_cost(call: ReadCall, delivered: tuple[int, int]) -> tuple[float, bool, bool]:
|
||||
lines = max(0, delivered[1] - delivered[0] + 1)
|
||||
line_tokens = call.token_per_line * lines
|
||||
# Approximate byte cap. The implementation scales byte cap as
|
||||
# max(50KiB, maxLinesToCollect * 512). For normal code line lengths this is
|
||||
# rarely binding; keep the indicator so huge-line configs are visible.
|
||||
byte_budget = max(CURRENT_MAX_BYTES, lines * 512)
|
||||
approx_bytes = line_tokens * 4
|
||||
bytes_limited = approx_bytes > byte_budget
|
||||
if bytes_limited:
|
||||
line_tokens = byte_budget / 4
|
||||
return call.arg_tokens + line_tokens, call.kind == "explicit" and lines >= call.config_max_lines if False else False, bytes_limited
|
||||
|
||||
|
||||
def load_reads(conn: sqlite3.Connection, since_ms: int) -> dict[tuple[str, str], list[ReadCall]]:
|
||||
sql = """
|
||||
SELECT c.session_file, c.seq, c.arg_json,
|
||||
COALESCE(c.arg_tokens,0), COALESCE(r.result_tokens,0)
|
||||
FROM ss_tool_calls c
|
||||
LEFT JOIN ss_tool_results r
|
||||
ON r.session_file = c.session_file
|
||||
AND r.call_id = c.call_id
|
||||
AND r.seq >= c.seq
|
||||
WHERE c.tool_name = 'read' AND c.timestamp >= ?
|
||||
ORDER BY c.session_file, c.seq
|
||||
"""
|
||||
groups: dict[tuple[str, str], list[ReadCall]] = defaultdict(list)
|
||||
for row in conn.execute(sql, (since_ms,)):
|
||||
call = parse_call(row)
|
||||
if call is None:
|
||||
continue
|
||||
groups[(call.session, call.file)].append(call)
|
||||
return groups
|
||||
|
||||
|
||||
def replay(groups: dict[tuple[str, str], list[ReadCall]], cfg: Config) -> ReplayResult:
|
||||
tokens = 0.0
|
||||
calls = 0
|
||||
skipped = 0
|
||||
trunc = 0
|
||||
bytes_limited = 0
|
||||
selector_tokens = 0.0
|
||||
selector_calls = 0
|
||||
selector_skipped = 0
|
||||
selector_groups = 0
|
||||
selector_groups_single_call = 0
|
||||
selector_groups_all_first = 0
|
||||
default_tokens = 0.0
|
||||
default_calls = 0
|
||||
default_skipped = 0
|
||||
raw_tokens = 0.0
|
||||
raw_calls = 0
|
||||
|
||||
for group in groups.values():
|
||||
first = group[0]
|
||||
selector_first = first.kind in ("explicit", "open")
|
||||
default_first = first.kind == "default"
|
||||
if selector_first:
|
||||
selector_groups += 1
|
||||
if len(group) == 1:
|
||||
selector_groups_single_call += 1
|
||||
coverage: list[tuple[int, int]] = []
|
||||
paid_selector_calls = 0
|
||||
covered_all_by_first = False
|
||||
|
||||
for idx, call in enumerate(group):
|
||||
req = requested_interval(call, cfg)
|
||||
delivered = delivered_interval(call, cfg)
|
||||
if req is None or delivered is None:
|
||||
cost = call.arg_tokens + call.result_tokens
|
||||
tokens += cost
|
||||
calls += 1
|
||||
raw_tokens += cost
|
||||
raw_calls += 1
|
||||
if selector_first:
|
||||
selector_tokens += cost
|
||||
selector_calls += 1
|
||||
paid_selector_calls += 1
|
||||
elif default_first:
|
||||
default_tokens += cost
|
||||
default_calls += 1
|
||||
continue
|
||||
|
||||
if idx > 0 and is_covered(coverage, req):
|
||||
skipped += 1
|
||||
if selector_first:
|
||||
selector_skipped += 1
|
||||
elif default_first:
|
||||
default_skipped += 1
|
||||
continue
|
||||
|
||||
lines = delivered[1] - delivered[0] + 1
|
||||
if lines >= cfg.max_lines and call.kind == "explicit":
|
||||
# Candidate max cap would truncate this explicit request.
|
||||
requested_len = max(1, (call.end or call.start or 1) - (call.start or 1) + 1)
|
||||
if requested_len + cfg.leading + cfg.trailing > cfg.max_lines:
|
||||
trunc += 1
|
||||
line_tokens = call.token_per_line * lines
|
||||
byte_budget = max(CURRENT_MAX_BYTES, lines * 512)
|
||||
if line_tokens * 4 > byte_budget:
|
||||
bytes_limited += 1
|
||||
line_tokens = byte_budget / 4
|
||||
cost = call.arg_tokens + line_tokens
|
||||
tokens += cost
|
||||
calls += 1
|
||||
if selector_first:
|
||||
selector_tokens += cost
|
||||
selector_calls += 1
|
||||
paid_selector_calls += 1
|
||||
elif default_first:
|
||||
default_tokens += cost
|
||||
default_calls += 1
|
||||
|
||||
coverage = add_interval(coverage, delivered)
|
||||
if idx == 0 and selector_first:
|
||||
# Check whether the first delivered interval covers every later
|
||||
# bounded request in the historical group.
|
||||
all_covered = True
|
||||
for later in group[1:]:
|
||||
later_req = requested_interval(later, cfg)
|
||||
if later_req is not None and not is_covered([delivered], later_req):
|
||||
all_covered = False
|
||||
break
|
||||
covered_all_by_first = all_covered
|
||||
|
||||
if selector_first and (covered_all_by_first or len(group) == 1):
|
||||
selector_groups_all_first += 1
|
||||
|
||||
return ReplayResult(
|
||||
config=cfg,
|
||||
tokens=tokens,
|
||||
calls=calls,
|
||||
skipped_calls=skipped,
|
||||
truncations=trunc,
|
||||
bytes_limited=bytes_limited,
|
||||
selector_tokens=selector_tokens,
|
||||
selector_calls=selector_calls,
|
||||
selector_skipped=selector_skipped,
|
||||
selector_groups=selector_groups,
|
||||
selector_groups_single_call=selector_groups_single_call,
|
||||
selector_groups_all_covered_by_first=selector_groups_all_first,
|
||||
default_tokens=default_tokens,
|
||||
default_calls=default_calls,
|
||||
default_skipped=default_skipped,
|
||||
raw_tokens=raw_tokens,
|
||||
raw_calls=raw_calls,
|
||||
)
|
||||
|
||||
|
||||
def candidate_grid(args) -> list[Config]:
|
||||
defaults = parse_int_list(args.defaults)
|
||||
maxes = parse_int_list(args.max_lines)
|
||||
leads = parse_int_list(args.leading)
|
||||
trails = parse_int_list(args.trailing)
|
||||
out: list[Config] = []
|
||||
for d in defaults:
|
||||
for m in maxes:
|
||||
if d > m:
|
||||
continue
|
||||
for l in leads:
|
||||
for t in trails:
|
||||
out.append(Config(d, m, l, t))
|
||||
return out
|
||||
|
||||
|
||||
def pareto(results: list[ReplayResult], max_truncations: int, max_regret_tokens: float = math.inf) -> list[ReplayResult]:
|
||||
# Frontier over (tokens lower, calls lower), excluding configs that truncate
|
||||
# more explicit requests than today's cap.
|
||||
clean = [r for r in results if r.truncations <= max_truncations and r.tokens <= max_regret_tokens]
|
||||
clean.sort(key=lambda r: (r.tokens, r.calls))
|
||||
frontier: list[ReplayResult] = []
|
||||
best_calls = math.inf
|
||||
for r in clean:
|
||||
if r.calls < best_calls:
|
||||
frontier.append(r)
|
||||
best_calls = r.calls
|
||||
return frontier
|
||||
|
||||
|
||||
def choose_recommended(results: list[ReplayResult], current: ReplayResult) -> ReplayResult:
|
||||
# Objective: minimize tokens plus a small penalty for still needing calls,
|
||||
# while requiring no *additional* explicit-request truncations and at least
|
||||
# current first-call coverage. One avoided read call is valued at ~250
|
||||
# tokens of ergonomics.
|
||||
viable = [
|
||||
r for r in results
|
||||
if r.truncations <= current.truncations
|
||||
and r.first_cover_rate >= current.first_cover_rate
|
||||
and r.tokens <= current.tokens * 1.02
|
||||
]
|
||||
if not viable:
|
||||
viable = [r for r in results if r.truncations <= current.truncations]
|
||||
if not viable:
|
||||
viable = results
|
||||
return min(viable, key=lambda r: r.tokens + 250 * r.calls + 100_000 * max(0, r.truncations - current.truncations))
|
||||
|
||||
|
||||
def print_result(prefix: str, r: ReplayResult, baseline: ReplayResult) -> None:
|
||||
dtok = r.tokens - baseline.tokens
|
||||
dcalls = r.calls - baseline.calls
|
||||
print(
|
||||
f"{prefix:<14} {r.config.label():<22} "
|
||||
f"tokens={r.tokens/1e6:8.2f}M ({dtok/baseline.tokens*100:+6.2f}%) "
|
||||
f"calls={r.calls:7,} ({dcalls:+7,}) "
|
||||
f"skipped={r.skipped_calls:6,} "
|
||||
f"first-cover={r.first_cover_rate*100:5.1f}% "
|
||||
f"trunc={r.truncations:4,}"
|
||||
)
|
||||
|
||||
|
||||
def plot(results: list[ReplayResult], current: ReplayResult, recommended: ReplayResult) -> Path:
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
plt.rcParams.update({"figure.dpi": 110, "font.size": 10})
|
||||
fig, axes = plt.subplots(2, 2, figsize=(15, 9))
|
||||
|
||||
xs = np.array([r.calls for r in results])
|
||||
ys = np.array([r.tokens / 1e6 for r in results])
|
||||
colors = np.array([r.config.default for r in results])
|
||||
sizes = np.array([20 + min(80, r.config.trailing * 5) for r in results])
|
||||
|
||||
ax = axes[0, 0]
|
||||
sc = ax.scatter(xs, ys, c=colors, s=sizes, cmap="viridis", alpha=0.65, edgecolors="none")
|
||||
ax.scatter([current.calls], [current.tokens / 1e6], marker="*", s=180, color="#111", label="current")
|
||||
ax.scatter([recommended.calls], [recommended.tokens / 1e6], marker="*", s=180, color="#dc2626", label="recommended")
|
||||
ax.set_xlabel("paid read calls after replay")
|
||||
ax.set_ylabel("estimated read tokens (M)")
|
||||
ax.set_title("candidate trade-off: tokens vs follow-up calls")
|
||||
ax.legend(frameon=False)
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
cbar = fig.colorbar(sc, ax=ax)
|
||||
cbar.set_label("defaultLimit")
|
||||
|
||||
ax = axes[0, 1]
|
||||
frontier = pareto(results, current.truncations)
|
||||
frontier.sort(key=lambda r: r.calls)
|
||||
ax.plot([r.calls for r in frontier], [r.tokens / 1e6 for r in frontier], color="#2563eb", linewidth=2)
|
||||
ax.scatter([r.calls for r in frontier], [r.tokens / 1e6 for r in frontier], color="#2563eb", s=20)
|
||||
ax.scatter([current.calls], [current.tokens / 1e6], marker="*", s=180, color="#111", label="current")
|
||||
ax.scatter([recommended.calls], [recommended.tokens / 1e6], marker="*", s=180, color="#dc2626", label="recommended")
|
||||
ax.set_xlabel("paid read calls")
|
||||
ax.set_ylabel("estimated read tokens (M)")
|
||||
ax.set_title("Pareto frontier (no extra explicit truncations)")
|
||||
ax.legend(frameon=False)
|
||||
ax.grid(True, alpha=0.25, linestyle="--")
|
||||
|
||||
ax = axes[1, 0]
|
||||
by_default: dict[int, list[ReplayResult]] = defaultdict(list)
|
||||
for r in results:
|
||||
if r.truncations <= current.truncations and r.config.leading == recommended.config.leading and r.config.trailing == recommended.config.trailing:
|
||||
by_default[r.config.default].append(r)
|
||||
defaults = sorted(by_default)
|
||||
vals = [min(v, key=lambda r: r.tokens).tokens / 1e6 for v in by_default.values()]
|
||||
ax.bar([str(d) for d in defaults], vals, color="#16a34a")
|
||||
ax.axhline(current.tokens / 1e6, color="#111", linestyle="--", linewidth=1, label="current")
|
||||
ax.set_xlabel("defaultLimit")
|
||||
ax.set_ylabel("best tokens (M)")
|
||||
ax.set_title(f"defaultLimit sensitivity (L={recommended.config.leading}, T={recommended.config.trailing})")
|
||||
ax.legend(frameon=False)
|
||||
ax.grid(True, axis="y", alpha=0.25, linestyle="--")
|
||||
|
||||
ax = axes[1, 1]
|
||||
top = sorted([r for r in results if r.truncations <= current.truncations], key=lambda r: r.tokens + 250 * r.calls)[:12]
|
||||
labels = [r.config.label() for r in top]
|
||||
token_delta = [(r.tokens - current.tokens) / current.tokens * 100 for r in top]
|
||||
call_delta = [(r.calls - current.calls) / current.calls * 100 for r in top]
|
||||
y = np.arange(len(top))
|
||||
ax.barh(y - 0.18, token_delta, height=0.35, color="#2563eb", label="token Δ%")
|
||||
ax.barh(y + 0.18, call_delta, height=0.35, color="#d97706", label="call Δ%")
|
||||
ax.set_yticks(y, labels)
|
||||
ax.invert_yaxis()
|
||||
ax.axvline(0, color="#111", linewidth=0.8)
|
||||
ax.set_xlabel("relative to current")
|
||||
ax.set_title("top configs by token+call objective")
|
||||
ax.legend(frameon=False)
|
||||
ax.grid(True, axis="x", alpha=0.25, linestyle="--")
|
||||
|
||||
fig.suptitle("read configuration counterfactual optimizer", fontsize=13, y=0.995)
|
||||
fig.tight_layout()
|
||||
out = OUT_DIR / "read-optimizer.png"
|
||||
fig.savefig(out, bbox_inches="tight")
|
||||
plt.close(fig)
|
||||
return out
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="read configuration optimizer")
|
||||
ap.add_argument("--since", default=DEFAULT_SINCE, help=f"YYYY-MM-DD (default {DEFAULT_SINCE})")
|
||||
ap.add_argument("--defaults", default="100,150,200,250,300,400,500,700,1000")
|
||||
ap.add_argument("--max-lines", default="500,750,1000,1500,2000,3000")
|
||||
ap.add_argument("--leading", default="0,3,5,10,20")
|
||||
ap.add_argument("--trailing", default="0,3,10,25,50,100,200")
|
||||
ap.add_argument("--top", type=int, default=15, help="print top N configs")
|
||||
args = ap.parse_args()
|
||||
|
||||
since = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
||||
since_ms = int(since.timestamp() * 1000)
|
||||
|
||||
if not DB_PATH.exists():
|
||||
sys.exit(f"db missing: {DB_PATH}")
|
||||
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
|
||||
groups = load_reads(conn, since_ms)
|
||||
conn.close()
|
||||
total_calls = sum(len(v) for v in groups.values())
|
||||
print(f"loaded {total_calls:,} read calls across {len(groups):,} (session,file) groups since {args.since}")
|
||||
|
||||
current = replay(groups, Config(CURRENT_DEFAULT, CURRENT_MAX_LINES, CURRENT_LEADING, CURRENT_TRAILING))
|
||||
configs = candidate_grid(args)
|
||||
# Ensure current is present even if user overrides grid.
|
||||
cur_cfg = Config(CURRENT_DEFAULT, CURRENT_MAX_LINES, CURRENT_LEADING, CURRENT_TRAILING)
|
||||
if cur_cfg not in configs:
|
||||
configs.append(cur_cfg)
|
||||
print(f"evaluating {len(configs):,} candidate configs")
|
||||
results = [replay(groups, cfg) for cfg in configs]
|
||||
recommended = choose_recommended(results, current)
|
||||
|
||||
print()
|
||||
print_result("current", current, current)
|
||||
print_result("recommended", recommended, current)
|
||||
|
||||
allowed = [r for r in results if r.truncations <= current.truncations]
|
||||
print(f"\nTop token-minimizing configs (truncations <= current {current.truncations:,}):")
|
||||
for i, r in enumerate(sorted(allowed, key=lambda r: r.tokens)[: args.top], 1):
|
||||
print_result(f"#{i}", r, current)
|
||||
|
||||
print(f"\nTop balanced configs (tokens + 250 tokens/read-call objective, truncations <= current {current.truncations:,}):")
|
||||
for i, r in enumerate(sorted(allowed, key=lambda r: r.tokens + 250 * r.calls)[: args.top], 1):
|
||||
print_result(f"#{i}", r, current)
|
||||
|
||||
no_call_increase = [r for r in allowed if r.calls <= current.calls]
|
||||
print(f"\nBest configs with calls <= current (truncations <= current {current.truncations:,}):")
|
||||
for i, r in enumerate(sorted(no_call_increase, key=lambda r: r.tokens)[: args.top], 1):
|
||||
print_result(f"#{i}", r, current)
|
||||
|
||||
print("\nRecommended breakdown:")
|
||||
print(f" selector groups : {recommended.selector_groups:,}")
|
||||
print(f" selector first-cover : {recommended.first_cover_rate*100:.1f}%")
|
||||
print(f" selector skipped calls : {recommended.selector_skipped:,}")
|
||||
print(f" default skipped calls : {recommended.default_skipped:,}")
|
||||
print(f" raw/unmodelled calls : {recommended.raw_calls:,}")
|
||||
print(f" byte-limited estimates : {recommended.bytes_limited:,}")
|
||||
|
||||
out = plot(results, current, recommended)
|
||||
print(f"\nwrote {out}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user