opencode-token-stats

#!/usr/bin/env python3
"""
Analyze token usage across recent OpenCode sessions.
Usage: opencode-token-stats [--sessions N] [--json]
"""
 
import json
import argparse
from pathlib import Path
from datetime import datetime
 
def analyze_sessions(num_sessions=10, output_json=False, session_id=None):
 storage = Path.home() / ".local/share/opencode/storage"
 message_dir = storage / "message"
 part_dir = storage / "part"
 session_dir = storage / "session"
 
 if not message_dir.exists():
 print("Error: OpenCode storage not found at", storage)
 return
 
 # Get sessions to analyze
 if session_id:
 # Analyze specific session
 session_path = message_dir / session_id
 if not session_path.exists():
 print(f"Error: Session {session_id} not found")
 return
 sessions = [session_path]
 else:
 # Get recent sessions sorted by modification time
 sessions = sorted(message_dir.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True)[:num_sessions]
 
 results = []
 grand_totals = {
 "input": 0, "output": 0, "reasoning": 0,
 "cache_read": 0, "cache_write": 0,
 "steps": 0, "sessions": 0,
 "reasons": {"tool-calls": 0, "stop": 0, "other": 0}
 }
 
 for session_path in sessions:
 session_id = session_path.name
 totals = {
 "input": 0, "output": 0, "reasoning": 0,
 "cache_read": 0, "cache_write": 0,
 "cost": 0.0, "steps": 0,
 "reasons": {"tool-calls": 0, "stop": 0, "other": 0}
 }
 
 # Get messages for this session
 msg_files = list(session_path.glob("*.json"))
 
 for msg_file in msg_files:
 msg_id = msg_file.stem
 parts_path = part_dir / msg_id
 if parts_path.exists():
 for part_file in parts_path.glob("*.json"):
 try:
 part = json.loads(part_file.read_text())
 if part.get("type") == "step-finish" and "tokens" in part:
 t = part["tokens"]
 totals["input"] += t.get("input", 0)
 totals["output"] += t.get("output", 0)
 totals["reasoning"] += t.get("reasoning", 0)
 cache = t.get("cache", {})
 totals["cache_read"] += cache.get("read", 0)
 totals["cache_write"] += cache.get("write", 0)
 totals["cost"] += part.get("cost", 0)
 totals["steps"] += 1
 
 reason = part.get("reason", "other")
 if reason in totals["reasons"]:
 totals["reasons"][reason] += 1
 else:
 totals["reasons"]["other"] += 1
 except (json.JSONDecodeError, KeyError):
 pass
 
 # Get session metadata (title, timestamps)
 title = "Unknown"
 created = None
 for s_dir in session_dir.iterdir():
 s_file = s_dir / f"{session_id}.json"
 if s_file.exists():
 try:
 sess = json.loads(s_file.read_text())
 title = sess.get("title", "Untitled")[:60]
 created = sess.get("createdAt")
 except (json.JSONDecodeError, KeyError):
 pass
 break
 
 # Calculate derived metrics
 total_tokens = totals["input"] + totals["output"] + totals["cache_read"]
 cache_hit_rate = (totals["cache_read"] / (totals["input"] + totals["cache_read"]) * 100) if (totals["input"] + totals["cache_read"]) > 0 else 0
 
 session_result = {
 "session_id": session_id,
 "title": title,
 "created": created,
 "steps": totals["steps"],
 "tokens": {
 "input": totals["input"],
 "output": totals["output"],
 "reasoning": totals["reasoning"],
 "cache_read": totals["cache_read"],
 "cache_write": totals["cache_write"],
 "total": total_tokens
 },
 "cost": totals["cost"],
 "cache_hit_rate": round(cache_hit_rate, 1),
 "finish_reasons": totals["reasons"]
 }
 results.append(session_result)
 
 # Update grand totals
 grand_totals["input"] += totals["input"]
 grand_totals["output"] += totals["output"]
 grand_totals["reasoning"] += totals["reasoning"]
 grand_totals["cache_read"] += totals["cache_read"]
 grand_totals["cache_write"] += totals["cache_write"]
 grand_totals["steps"] += totals["steps"]
 grand_totals["sessions"] += 1
 for reason, count in totals["reasons"].items():
 grand_totals["reasons"][reason] += count
 
 # Output
 if output_json:
 output = {
 "sessions": results,
 "totals": grand_totals,
 "generated_at": datetime.now().isoformat()
 }
 print(json.dumps(output, indent=2))
 else:
 print_summary(results, grand_totals)
 
def print_summary(results, grand_totals):
 print("=" * 120)
 print("OPENCODE SESSION TOKEN ANALYSIS")
 print("=" * 120)
 print()
 
 # Per-session breakdown
 print(f"{'Session':<25} {'Title':<30} {'Steps':>6} {'Input':>12} {'Output':>10} {'Reasoning':>10} {'Cache Read':>12} {'Cache Write':>12} {'Cache %':>8}")
 print("-" * 120)
 
 for r in results:
 t = r["tokens"]
 print(f"{r['session_id'][:24]:<25} {r['title'][:29]:<30} {r['steps']:>6} {t['input']:>12,} {t['output']:>10,} {t['reasoning']:>10,} {t['cache_read']:>12,} {t['cache_write']:>12,} {r['cache_hit_rate']:>7.1f}%")
 
 print("-" * 120)
 print()
 
 # Grand totals
 total_all = grand_totals["input"] + grand_totals["output"] + grand_totals["cache_read"]
 overall_cache_rate = (grand_totals["cache_read"] / (grand_totals["input"] + grand_totals["cache_read"]) * 100) if (grand_totals["input"] + grand_totals["cache_read"]) > 0 else 0
 
 print("TOTALS ACROSS ALL SESSIONS")
 print("-" * 50)
 print(f" Sessions analyzed: {grand_totals['sessions']:>15,}")
 print(f" Total steps: {grand_totals['steps']:>15,}")
 avg_steps = grand_totals['steps'] / grand_totals['sessions'] if grand_totals['sessions'] > 0 else 0
 print(f" Avg steps/session: {avg_steps:>15.1f}")
 print()
 print(" TOKEN BREAKDOWN:")
 print(f" Input tokens: {grand_totals['input']:>15,}")
 print(f" Output tokens: {grand_totals['output']:>15,}")
 print(f" Reasoning tokens: {grand_totals['reasoning']:>15,}")
 print(f" Cache read: {grand_totals['cache_read']:>15,}")
 print(f" Cache write: {grand_totals['cache_write']:>15,}")
 print(f" ─────────────────────────────────────────────")
 print(f" TOTAL: {total_all:>15,}")
 print()
 print(f" Overall cache hit rate: {overall_cache_rate:.1f}%")
 print()
 print(" STEP FINISH REASONS:")
 for reason, count in grand_totals["reasons"].items():
 if count > 0:
 print(f" {reason}: {count:>10,}")
 print()
 print("=" * 120)
 
def main():
 parser = argparse.ArgumentParser(description="Analyze OpenCode session token usage")
 parser.add_argument("--sessions", "-n", type=int, default=10, help="Number of recent sessions to analyze (default: 10)")
 parser.add_argument("--session", "-s", type=str, default=None, help="Analyze specific session ID")
 parser.add_argument("--json", "-j", action="store_true", help="Output as JSON instead of formatted text")
 args = parser.parse_args()
 
 analyze_sessions(num_sessions=args.sessions, output_json=args.json, session_id=args.session)
 
if __name__ == "__main__":
 main()