#!/usr/bin/env python3 """Deterministic session analytics for the /insights command. Walks ~/.pi/agent/sessions, computes usage stats over a window of days, and prints a JSON summary on stdout. No LLM involved here. """ import sys import os import json import glob import collections from datetime import datetime, timezone, timedelta SESSIONS_DIR = os.path.expanduser("~/.pi/agent/sessions") def parse_ts(s): if not s: return None try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: return None def project_name(cwd): if not cwd: return "(unknown)" return os.path.basename(cwd.rstrip("/")) or cwd def analyze(days=30): cutoff = datetime.now(timezone.utc) - timedelta(days=days) files = glob.glob(os.path.join(SESSIONS_DIR, "*", "*.jsonl")) sessions = [] tool_counts = collections.Counter() model_counts = collections.Counter() project_stats = collections.defaultdict(lambda: { "sessions": 0, "user_msgs": 0, "assistant_msgs": 0, "tool_calls": 0, "cost": 0.0, "tokens": 0, }) hour_hist = collections.Counter() dow_hist = collections.Counter() day_hist = collections.Counter() totals = { "sessions": 0, "user_msgs": 0, "assistant_msgs": 0, "tool_results": 0, "tool_calls": 0, "cost": 0.0, "input_tokens": 0, "output_tokens": 0, "cache_read": 0, "cache_write": 0, "total_tokens": 0, "compactions": 0, "branches": 0, } for fpath in files: header = None first_ts = None last_ts = None s_user = s_asst = s_tools = s_toolres = 0 s_cost = 0.0 s_tokens = 0 s_compactions = 0 s_branches = 0 s_models = set() s_tool_counter = collections.Counter() cwd = None try: with open(fpath, "r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: e = json.loads(line) except Exception: continue et = e.get("type") if et == "session": header = e cwd = e.get("cwd") first_ts = parse_ts(e.get("timestamp")) continue ts = parse_ts(e.get("timestamp")) if ts: if first_ts is None: first_ts = ts last_ts = ts if et == "compaction": s_compactions += 1 elif et == "branch_summary": s_branches += 1 elif et == "message": msg = e.get("message", {}) role = msg.get("role") if role == "user": s_user += 1 elif role == "assistant": s_asst += 1 if msg.get("model"): s_models.add(msg["model"]) u = msg.get("usage") or {} s_cost += (u.get("cost") or {}).get("total", 0) or 0 s_tokens += u.get("totalTokens", 0) or 0 totals["input_tokens"] += u.get("input", 0) or 0 totals["output_tokens"] += u.get("output", 0) or 0 totals["cache_read"] += u.get("cacheRead", 0) or 0 totals["cache_write"] += u.get("cacheWrite", 0) or 0 for c in msg.get("content", []): if isinstance(c, dict) and c.get("type") == "toolCall": name = c.get("name", "?") s_tools += 1 s_tool_counter[name] += 1 elif role == "toolResult": s_toolres += 1 except Exception: continue # Window filter: use session start time anchor = first_ts or last_ts if anchor is None or anchor < cutoff: continue proj = project_name(cwd) ps = project_stats[proj] ps["sessions"] += 1 ps["user_msgs"] += s_user ps["assistant_msgs"] += s_asst ps["tool_calls"] += s_tools ps["cost"] += s_cost ps["tokens"] += s_tokens for name, n in s_tool_counter.items(): tool_counts[name] += n for m in s_models: model_counts[m] += 1 if anchor: local = anchor.astimezone() hour_hist[local.hour] += 1 dow_hist[local.weekday()] += 1 day_hist[local.strftime("%Y-%m-%d")] += 1 duration_min = None if first_ts and last_ts and last_ts > first_ts: duration_min = round((last_ts - first_ts).total_seconds() / 60.0, 1) sessions.append({ "file": os.path.basename(fpath), "project": proj, "cwd": cwd, "start": anchor.isoformat() if anchor else None, "duration_min": duration_min, "user_msgs": s_user, "assistant_msgs": s_asst, "tool_calls": s_tools, "tool_results": s_toolres, "cost": round(s_cost, 4), "tokens": s_tokens, "compactions": s_compactions, "branches": s_branches, "models": sorted(s_models), }) totals["sessions"] += 1 totals["user_msgs"] += s_user totals["assistant_msgs"] += s_asst totals["tool_results"] += s_toolres totals["tool_calls"] += s_tools totals["cost"] += s_cost totals["total_tokens"] += s_tokens totals["compactions"] += s_compactions totals["branches"] += s_branches totals["cost"] = round(totals["cost"], 4) projects = [] for name, ps in project_stats.items(): projects.append({ "project": name, "sessions": ps["sessions"], "user_msgs": ps["user_msgs"], "assistant_msgs": ps["assistant_msgs"], "tool_calls": ps["tool_calls"], "cost": round(ps["cost"], 4), "tokens": ps["tokens"], }) projects.sort(key=lambda p: p["sessions"], reverse=True) durations = [s["duration_min"] for s in sessions if s["duration_min"]] avg_duration = round(sum(durations) / len(durations), 1) if durations else None active_days = len(day_hist) return { "window_days": days, "generated_at": datetime.now(timezone.utc).isoformat(), "totals": totals, "active_days": active_days, "avg_session_duration_min": avg_duration, "projects": projects, "tool_usage": [{"name": k, "count": v} for k, v in tool_counts.most_common()], "model_usage": [{"model": k, "sessions": v} for k, v in model_counts.most_common()], "hour_histogram": [hour_hist.get(h, 0) for h in range(24)], "dow_histogram": [dow_hist.get(d, 0) for d in range(7)], "daily_histogram": dict(sorted(day_hist.items())), "longest_sessions": sorted( [s for s in sessions if s["duration_min"]], key=lambda s: s["duration_min"], reverse=True )[:10], "costliest_sessions": sorted(sessions, key=lambda s: s["cost"], reverse=True)[:10], } if __name__ == "__main__": days = 30 if len(sys.argv) > 1: try: days = int(sys.argv[1]) except ValueError: pass print(json.dumps(analyze(days), indent=2))