From bc75b13873da81fb1c3db83e4c17680696e55ddc Mon Sep 17 00:00:00 2001 From: amnesiaof Date: Fri, 2 Oct 2026 18:26:57 +0300 Subject: [PATCH] chore(scripts): remove unreferenced Python analysis CLIs Removes seven extension-less Python CLIs and the shared opencode_api.py module, plus tests/test-dcp-cache.sh, which was their only caller. Nothing reached them. No npm script, no CI step, no doc. The files array never published scripts/, so they shipped to no one who installed the package. scripts/ is also listed in .repomixignore. Deleted with git grep confirming the only remaining references are in package-lock.json and the files being removed. --- scripts/opencode-dcp-stats | 452 -------------------------- scripts/opencode-find-session | 132 -------- scripts/opencode-get-message | 98 ------ scripts/opencode-message-token-counts | 393 ---------------------- scripts/opencode-session-timeline | 258 --------------- scripts/opencode-token-stats | 182 ----------- scripts/opencode_api.py | 249 -------------- tests/test-dcp-cache.sh | 364 --------------------- 8 files changed, 2128 deletions(-) delete mode 100755 scripts/opencode-dcp-stats delete mode 100755 scripts/opencode-find-session delete mode 100755 scripts/opencode-get-message delete mode 100755 scripts/opencode-message-token-counts delete mode 100755 scripts/opencode-session-timeline delete mode 100755 scripts/opencode-token-stats delete mode 100644 scripts/opencode_api.py delete mode 100755 tests/test-dcp-cache.sh diff --git a/scripts/opencode-dcp-stats b/scripts/opencode-dcp-stats deleted file mode 100755 index 44a4322f..00000000 --- a/scripts/opencode-dcp-stats +++ /dev/null @@ -1,452 +0,0 @@ -#!/usr/bin/env python3 -""" -Analyze Dynamic Context Pruning (DCP) tool impact on cache efficiency. -Tracks cache hit rates and context size changes before/after DCP tool invocations. - -Usage: opencode-dcp-stats [--sessions N] [--min-messages M] [--json] [--verbose] -""" - -import json -import argparse -from datetime import datetime -from collections import defaultdict -from typing import Optional - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - -# DCP tool names across versions (compress is canonical; others are legacy aliases) -DCP_TOOLS = { - "compress", "prune", "distill", - "discard", "extract", "context_pruning", "squash", "consolidate" -} - -# Anthropic pricing: cache read is ~10% of input cost -CACHE_READ_COST_PER_1K = 0.00030 # $0.30 per 1M tokens -INPUT_COST_PER_1K = 0.003 # $3.00 per 1M tokens - - -def get_session_messages(client, session: dict) -> list[dict]: - """Get all messages for a session.""" - return client.get_session_messages(session["id"], directory=session.get("directory")) - - -def is_ignored_message(message: dict, parts: list[dict]) -> bool: - """ - Check if a message should be ignored (DCP notification messages). - Returns True if message has no parts OR all parts have ignored=true. - Mirrors the isIgnoredUserMessage logic from the DCP plugin. - """ - if not parts: - return True - - # Check text parts for ignored flag - text_parts = [p for p in parts if p.get("type") == "text"] - if not text_parts: - return False - - for part in text_parts: - if not part.get("ignored", False): - return False - - return True - - -def count_real_user_messages(messages: list[dict]) -> int: - """Count user messages that are not ignored (real user interactions).""" - count = 0 - - for msg in messages: - info = msg.get("info", {}) - parts = msg.get("parts", []) - # Only count user role messages - if info.get("role") != "user": - continue - - if not is_ignored_message(msg, parts): - count += 1 - - return count - - -def extract_step_finish(parts: list[dict]) -> Optional[dict]: - """Extract step-finish record from message parts.""" - for part in parts: - if part.get("type") == "step-finish" and "tokens" in part: - return part - return None - - -def extract_dcp_tools(parts: list[dict]) -> list[dict]: - """Extract all DCP tool calls from message parts.""" - dcp_calls = [] - for part in parts: - if part.get("type") == "tool": - tool_name = part.get("tool", "") - if tool_name in DCP_TOOLS: - dcp_calls.append({ - "tool": tool_name, - "state": part.get("state", {}), - "part_id": part.get("_id", "") - }) - return dcp_calls - - -def calc_cache_hit_rate(tokens: dict) -> float: - """Calculate cache hit rate from token dict.""" - input_tokens = tokens.get("input", 0) - cache = tokens.get("cache", {}) - cache_read = cache.get("read", 0) - total_context = input_tokens + cache_read - if total_context == 0: - return 0.0 - return (cache_read / total_context) * 100 - - -def analyze_session(messages: list[dict], session_id: str) -> dict: - """Analyze DCP impact for a single session.""" - - result = { - "session_id": session_id, - "dcp_events": [], - "total_dcp_calls": 0, - "total_steps": 0, - "by_tool": defaultdict(lambda: { - "calls": 0, - "hit_rate_before_sum": 0, - "hit_rate_after_sum": 0, - "context_before_sum": 0, - "context_after_sum": 0, - "input_before_sum": 0, - "input_after_sum": 0, - "cache_before_sum": 0, - "cache_after_sum": 0, - "events_with_data": 0 - }), - # Track hit rates by distance from last DCP call - "hit_rates_by_distance": defaultdict(list) - } - - prev_step = None - prev_dcp_tools = [] - steps_since_dcp = None # None = no DCP yet, 0 = just had DCP, 1+ = steps after - - for i, msg in enumerate(messages): - msg_info = msg.get("info", {}) - msg_id = msg_info.get("id", "") - parts = msg.get("parts", []) - - step_finish = extract_step_finish(parts) - dcp_tools = extract_dcp_tools(parts) - - if step_finish: - result["total_steps"] += 1 - tokens = step_finish.get("tokens", {}) - curr_hit_rate = calc_cache_hit_rate(tokens) - - # Track hit rate by distance from last DCP call - if steps_since_dcp is not None: - result["hit_rates_by_distance"][steps_since_dcp].append(curr_hit_rate) - steps_since_dcp += 1 - - # If previous step had DCP tools, measure impact - if prev_dcp_tools and prev_step is not None: - prev_tokens = prev_step.get("tokens", {}) - - prev_input = prev_tokens.get("input", 0) - prev_cache = prev_tokens.get("cache", {}).get("read", 0) - prev_context = prev_input + prev_cache - prev_hit_rate = calc_cache_hit_rate(prev_tokens) - - curr_input = tokens.get("input", 0) - curr_cache = tokens.get("cache", {}).get("read", 0) - curr_context = curr_input + curr_cache - curr_hit_rate = calc_cache_hit_rate(tokens) - - for dcp in prev_dcp_tools: - tool_name = dcp["tool"] - result["total_dcp_calls"] += 1 - - event = { - "tool": tool_name, - "input_before": prev_input, - "input_after": curr_input, - "cache_before": prev_cache, - "cache_after": curr_cache, - "context_before": prev_context, - "context_after": curr_context, - "hit_rate_before": round(prev_hit_rate, 1), - "hit_rate_after": round(curr_hit_rate, 1), - "hit_rate_delta": round(curr_hit_rate - prev_hit_rate, 1), - "context_delta": curr_context - prev_context, - "message_id": msg_id - } - result["dcp_events"].append(event) - - # Aggregate stats - stats = result["by_tool"][tool_name] - stats["calls"] += 1 - stats["hit_rate_before_sum"] += prev_hit_rate - stats["hit_rate_after_sum"] += curr_hit_rate - stats["context_before_sum"] += prev_context - stats["context_after_sum"] += curr_context - stats["input_before_sum"] += prev_input - stats["input_after_sum"] += curr_input - stats["cache_before_sum"] += prev_cache - stats["cache_after_sum"] += curr_cache - stats["events_with_data"] += 1 - - prev_step = step_finish - prev_dcp_tools = dcp_tools - - # Reset distance counter if this step had DCP tools - if dcp_tools: - steps_since_dcp = 0 - - return result - - -def analyze_sessions( - client, - num_sessions: int = 20, - min_messages: int = 5, - output_json: bool = False, - verbose: bool = False, - session_id: str = None, - session_list_limit: int = 5000, -): - """Analyze DCP impact across recent sessions.""" - - # Get sessions to analyze - if session_id: - # Analyze specific session - sessions = [client.get_session(session_id)] - else: - sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit)[:num_sessions] - - all_results = [] - grand_totals = { - "sessions_analyzed": 0, - "sessions_with_dcp": 0, - "sessions_skipped_short": 0, - "total_dcp_calls": 0, - "total_steps": 0, - "min_messages_filter": min_messages, - "by_tool": defaultdict(lambda: { - "calls": 0, - "hit_rate_before_sum": 0, - "hit_rate_after_sum": 0, - "context_before_sum": 0, - "context_after_sum": 0, - "input_before_sum": 0, - "input_after_sum": 0, - "cache_before_sum": 0, - "cache_after_sum": 0, - "events_with_data": 0 - }), - "hit_rates_by_distance": defaultdict(list) - } - - for session in sessions: - session_id = session.get("id", "") - messages = get_session_messages(client, session) - - # Check minimum message count (excluding ignored messages) - real_user_messages = count_real_user_messages(messages) - if real_user_messages < min_messages: - grand_totals["sessions_skipped_short"] += 1 - continue - - result = analyze_session(messages, session_id) - result["user_messages"] = real_user_messages - - # Session metadata from API - result["title"] = session.get("title", "Untitled")[:50] - - if result["total_dcp_calls"] > 0: - all_results.append(result) - grand_totals["sessions_with_dcp"] += 1 - - grand_totals["sessions_analyzed"] += 1 - grand_totals["total_dcp_calls"] += result["total_dcp_calls"] - grand_totals["total_steps"] += result["total_steps"] - - for tool, stats in result["by_tool"].items(): - for key in stats: - grand_totals["by_tool"][tool][key] += stats[key] - - # Aggregate hit rates by distance - for dist, rates in result["hit_rates_by_distance"].items(): - grand_totals["hit_rates_by_distance"][dist].extend(rates) - - if output_json: - output = { - "sessions": all_results, - "totals": dict(grand_totals), - "generated_at": datetime.now().isoformat() - } - output["totals"]["by_tool"] = {k: dict(v) for k, v in grand_totals["by_tool"].items()} - print(json.dumps(output, indent=2, default=str)) - else: - print_summary(all_results, grand_totals, verbose) - - -def print_summary(results: list, totals: dict, verbose: bool = False): - """Print human-readable summary.""" - print("=" * 110) - print("DCP (Dynamic Context Pruning) Cache Impact Analysis") - print("=" * 110) - print() - - print("OVERVIEW") - print("-" * 50) - print(f" Min user messages filter: {totals['min_messages_filter']:>10,}") - print(f" Sessions scanned: {totals['sessions_analyzed'] + totals['sessions_skipped_short']:>10,}") - print(f" Sessions skipped (short): {totals['sessions_skipped_short']:>10,}") - print(f" Sessions analyzed: {totals['sessions_analyzed']:>10,}") - print(f" Sessions with DCP: {totals['sessions_with_dcp']:>10,}") - print(f" Total DCP tool calls: {totals['total_dcp_calls']:>10,}") - print(f" Total steps: {totals['total_steps']:>10,}") - print() - - # Per-tool breakdown with cache hit rate and context changes - if totals["by_tool"]: - print("PER-TOOL BREAKDOWN") - print("-" * 110) - print(f"{'Tool':<18} {'Calls':>7} {'Avg Hit% Before':>16} {'Avg Hit% After':>15} {'Delta':>8} {'Avg Ctx Before':>15} {'Avg Ctx After':>14}") - print("-" * 110) - - for tool, stats in sorted(totals["by_tool"].items(), key=lambda x: x[1]["calls"], reverse=True): - calls = stats["calls"] - if calls == 0: - continue - - avg_hit_before = stats["hit_rate_before_sum"] / calls - avg_hit_after = stats["hit_rate_after_sum"] / calls - hit_delta = avg_hit_after - avg_hit_before - avg_ctx_before = stats["context_before_sum"] / calls - avg_ctx_after = stats["context_after_sum"] / calls - - delta_str = f"{hit_delta:+.1f}%" - print(f"{tool:<18} {calls:>7,} {avg_hit_before:>15.1f}% {avg_hit_after:>14.1f}% {delta_str:>8} {avg_ctx_before:>14,.0f} {avg_ctx_after:>13,.0f}") - print() - - # Cost analysis - print("COST IMPACT ANALYSIS") - print("-" * 80) - total_input_before = sum(s["input_before_sum"] for s in totals["by_tool"].values()) - total_input_after = sum(s["input_after_sum"] for s in totals["by_tool"].values()) - total_cache_before = sum(s["cache_before_sum"] for s in totals["by_tool"].values()) - total_cache_after = sum(s["cache_after_sum"] for s in totals["by_tool"].values()) - - # Tokens moved from cache to input = cache lost - cache_preserved = total_cache_after - cache_lost_to_input = max(0, total_cache_before - total_cache_after) - - if totals["total_dcp_calls"] > 0: - # Average context reduction - avg_ctx_before = (total_input_before + total_cache_before) / totals["total_dcp_calls"] - avg_ctx_after = (total_input_after + total_cache_after) / totals["total_dcp_calls"] - ctx_reduction = avg_ctx_before - avg_ctx_after - - print(f" Total DCP events measured: {totals['total_dcp_calls']:>12,}") - print(f" Avg context before DCP: {avg_ctx_before:>12,.0f} tokens") - print(f" Avg context after DCP: {avg_ctx_after:>12,.0f} tokens") - print(f" Avg context change: {ctx_reduction:>+12,.0f} tokens") - print() - - # Cache efficiency - overall_hit_before = (total_cache_before / (total_input_before + total_cache_before) * 100) if (total_input_before + total_cache_before) > 0 else 0 - overall_hit_after = (total_cache_after / (total_input_after + total_cache_after) * 100) if (total_input_after + total_cache_after) > 0 else 0 - - print(f" Overall cache hit rate before: {overall_hit_before:>11.1f}%") - print(f" Overall cache hit rate after: {overall_hit_after:>11.1f}%") - print(f" Hit rate change: {overall_hit_after - overall_hit_before:>+11.1f}%") - print() - - # Cache recovery analysis - hit rates by distance from last DCP call - if totals["hit_rates_by_distance"]: - print("CACHE RECOVERY ANALYSIS (Hit Rate by Steps Since Last DCP Call)") - print("-" * 80) - print(f"{'Steps After DCP':<18} {'Samples':>10} {'Avg Hit%':>12} {'Min':>10} {'Max':>10}") - print("-" * 80) - - for dist in sorted(totals["hit_rates_by_distance"].keys())[:15]: - rates = totals["hit_rates_by_distance"][dist] - if rates: - avg_rate = sum(rates) / len(rates) - min_rate = min(rates) - max_rate = max(rates) - print(f"{dist:<18} {len(rates):>10,} {avg_rate:>11.1f}% {min_rate:>9.1f}% {max_rate:>9.1f}%") - print() - print(" (If cache fully recovers, later steps should approach 85%)") - print() - - # Per-session breakdown (if verbose or few sessions) - if verbose or len(results) <= 10: - if results: - print("PER-SESSION BREAKDOWN") - print("-" * 110) - print(f"{'Session':<25} {'Title':<30} {'User Msgs':>10} {'DCP Calls':>10} {'Avg Hit% Delta':>15}") - print("-" * 110) - - for r in results: - sid = r["session_id"][:24] - title = r["title"][:29] - user_msgs = r.get("user_messages", 0) - avg_delta = 0 - if r["dcp_events"]: - avg_delta = sum(e["hit_rate_delta"] for e in r["dcp_events"]) / len(r["dcp_events"]) - delta_str = f"{avg_delta:+.1f}%" - print(f"{sid:<25} {title:<30} {user_msgs:>10,} {r['total_dcp_calls']:>10,} {delta_str:>15}") - print() - - # Individual DCP events (only in verbose mode) - if verbose: - print("INDIVIDUAL DCP EVENTS") - print("-" * 110) - for r in results: - if r["dcp_events"]: - print(f"\n Session: {r['title'][:60]}") - for event in r["dcp_events"][:15]: - ctx_delta = f"{event['context_delta']:+,}" - hit_delta = f"{event['hit_rate_delta']:+.1f}%" - print(f" {event['tool']:<12} hit%: {event['hit_rate_before']:>5.1f} -> {event['hit_rate_after']:>5.1f} ({hit_delta:>7}) ctx: {event['context_before']:>8,} -> {event['context_after']:>8,} ({ctx_delta:>8})") - - print("=" * 110) - - -def main(): - parser = argparse.ArgumentParser(description="Analyze DCP tool cache impact") - parser.add_argument("--sessions", "-n", type=int, default=20, - help="Number of recent sessions to scan (default: 20)") - parser.add_argument("--session", "-s", type=str, default=None, - help="Analyze specific session ID") - parser.add_argument("--min-messages", "-m", type=int, default=5, - help="Minimum real user messages required (default: 5)") - parser.add_argument("--json", "-j", action="store_true", - help="Output as JSON") - parser.add_argument("--verbose", "-v", action="store_true", - help="Show detailed per-event breakdown") - add_api_arguments(parser) - args = parser.parse_args() - - try: - with create_client_from_args(args) as client: - analyze_sessions( - client, - num_sessions=args.sessions, - min_messages=args.min_messages, - output_json=args.json, - verbose=args.verbose, - session_id=args.session, - session_list_limit=args.session_list_limit, - ) - except APIError as err: - print(f"Error: {err}") - return 1 - - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/opencode-find-session b/scripts/opencode-find-session deleted file mode 100755 index 9e5f3de2..00000000 --- a/scripts/opencode-find-session +++ /dev/null @@ -1,132 +0,0 @@ -#!/usr/bin/env python3 -""" -Find OpenCode session IDs by title search using the OpenCode database. -Returns matching session IDs ordered by last usage time. - -Usage: opencode-find-session [--exact] [--json] -""" - -import json -import argparse -from datetime import datetime - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - - -def get_all_sessions(client, session_list_limit: int) -> list[dict]: - """Get all sessions with normalized metadata.""" - api_sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit) - sessions = [] - for session in api_sessions: - time_data = session.get("time", {}) - updated_ms = time_data.get("updated") or time_data.get("created") or 0 - last_used = updated_ms / 1000 if updated_ms else 0 - sessions.append( - { - "id": session.get("id", ""), - "title": session.get("title", "Untitled"), - "created_at": time_data.get("created"), - "last_used": last_used, - "last_used_iso": datetime.fromtimestamp(last_used).isoformat() if last_used else None, - } - ) - return sessions - - -def search_sessions(sessions: list[dict], search_term: str, exact: bool = False) -> list[dict]: - """Search sessions by title.""" - results = [] - search_lower = search_term.lower() - - for session in sessions: - title = session.get("title", "") - title_lower = title.lower() - - if exact: - if title_lower == search_lower: - results.append(session) - else: - if search_lower in title_lower: - results.append(session) - - # Sort by last used time, most recent first - results.sort(key=lambda s: s["last_used"], reverse=True) - - return results - - -def print_results(results: list[dict], search_term: str): - """Print search results.""" - if not results: - print(f"No sessions found matching: {search_term}") - return - - if len(results) == 1: - # Single result - just print the ID for easy piping - print(results[0]["id"]) - else: - # Multiple results - show a table - print(f"Found {len(results)} sessions matching: {search_term}") - print() - print(f"{'Session ID':<32} {'Last Used':<20} {'Title'}") - print("-" * 100) - - for r in results: - last_used = datetime.fromtimestamp(r["last_used"]).strftime("%Y-%m-%d %H:%M") - title = r["title"][:50] if len(r["title"]) > 50 else r["title"] - print(f"{r['id']:<32} {last_used:<20} {title}") - - -def main(): - parser = argparse.ArgumentParser( - description="Find OpenCode session IDs by title search" - ) - parser.add_argument( - "search_term", - nargs="?", - type=str, - help="Text to search for in session titles" - ) - parser.add_argument( - "--exact", "-e", - action="store_true", - help="Require exact title match (case-insensitive)" - ) - parser.add_argument( - "--json", "-j", - action="store_true", - help="Output as JSON" - ) - parser.add_argument( - "--all", "-a", - action="store_true", - help="Show all sessions (ignore search term)" - ) - add_api_arguments(parser) - args = parser.parse_args() - - if not args.all and not args.search_term: - parser.error("search_term is required unless --all is used") - - try: - with create_client_from_args(args) as client: - sessions = get_all_sessions(client, args.session_list_limit) - except APIError as err: - print(f"Error: {err}") - return 1 - - if args.all: - results = sorted(sessions, key=lambda s: s["last_used"], reverse=True) - else: - results = search_sessions(sessions, args.search_term, args.exact) - - if args.json: - print(json.dumps(results, indent=2, default=str)) - else: - print_results(results, args.search_term if not args.all else "(all)") - - return 0 - - -if __name__ == "__main__": - exit(main()) diff --git a/scripts/opencode-get-message b/scripts/opencode-get-message deleted file mode 100755 index ac39a623..00000000 --- a/scripts/opencode-get-message +++ /dev/null @@ -1,98 +0,0 @@ -#!/usr/bin/env python3 -""" -Get full OpenCode message payload(s) by message ID from the OpenCode database. - -Usage: - opencode-get-message [message-id ...] - opencode-get-message --session [message-id ...] -""" - -import argparse -import json - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - - -def normalize_message_payload(payload: dict) -> dict: - """Normalize API response into expected output shape.""" - info = payload.get("info", {}) - parts = payload.get("parts", []) - return {"info": info, "parts": parts} - - -def not_found_result(message_id: str) -> dict: - return {"id": message_id, "error": "message_not_found"} - - -def get_message_for_session(client, session: dict, message_id: str) -> dict: - """Get a message payload within a known session.""" - session_id = session.get("id", "") - directory = session.get("directory") - try: - payload = client.get_session_message(session_id, message_id, directory=directory) - return normalize_message_payload(payload) - except APIError as err: - if err.status_code == 404: - return not_found_result(message_id) - raise - - -def find_messages_without_session(client, message_ids: list[str], scan_sessions: int, session_list_limit: int) -> list[dict]: - """Search recent sessions for requested message IDs.""" - wanted = set(message_ids) - found: dict[str, dict] = {} - - sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit) - if scan_sessions > 0: - sessions = sessions[:scan_sessions] - - for session in sessions: - if not wanted: - break - messages = client.get_session_messages(session.get("id", ""), directory=session.get("directory")) - for message in messages: - info = message.get("info", {}) - mid = info.get("id") - if mid in wanted: - found[mid] = normalize_message_payload(message) - wanted.remove(mid) - - return [found.get(message_id, not_found_result(message_id)) for message_id in message_ids] - - -def main() -> int: - parser = argparse.ArgumentParser(description="Get full OpenCode message payload by message ID") - parser.add_argument("--session", "-s", type=str, default=None, help="Session ID for direct lookup") - parser.add_argument( - "--scan-sessions", - type=int, - default=200, - help="When --session is omitted, scan this many recent sessions for message IDs", - ) - parser.add_argument("message_ids", nargs="+", help="One or more message IDs") - add_api_arguments(parser) - args = parser.parse_args() - - try: - with create_client_from_args(args) as client: - if args.session: - session = client.get_session(args.session) - results = [get_message_for_session(client, session, message_id) for message_id in args.message_ids] - else: - results = find_messages_without_session( - client, - args.message_ids, - scan_sessions=args.scan_sessions, - session_list_limit=args.session_list_limit, - ) - except APIError as err: - print(f"Error: {err}") - return 1 - - output = results[0] if len(results) == 1 else results - print(json.dumps(output, indent=2)) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/opencode-message-token-counts b/scripts/opencode-message-token-counts deleted file mode 100755 index b04022f5..00000000 --- a/scripts/opencode-message-token-counts +++ /dev/null @@ -1,393 +0,0 @@ -#!/usr/bin/env python3 -""" -Show countAllMessageTokens-style token counts for each message in an OpenCode session. - -Usage: opencode-message-token-counts [--session ID] [--json] [--no-color] [--db PATH] -""" - -from __future__ import annotations - -import argparse -import json -import shutil -import subprocess -from pathlib import Path -from typing import Optional - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - - -SCRIPT_DIR = Path(__file__).resolve().parent -REPO_ROOT = SCRIPT_DIR.parent - - -class Colors: - RESET = "\033[0m" - BOLD = "\033[1m" - DIM = "\033[2m" - GREEN = "\033[32m" - YELLOW = "\033[33m" - CYAN = "\033[36m" - - -NO_COLOR = Colors() -for attr in dir(NO_COLOR): - if not attr.startswith("_"): - setattr(NO_COLOR, attr, "") - - -def stringify_json(value) -> str: - return json.dumps(value, separators=(",", ":"), ensure_ascii=False) - - -def collapse_whitespace(text: str) -> str: - return " ".join(text.split()) - - -def truncate(text: str, limit: int = 64) -> str: - if len(text) <= limit: - return text - return text[: limit - 3] + "..." - - -def get_terminal_width(default: int = 120) -> int: - return max(80, shutil.get_terminal_size((default, 20)).columns) - - -def short_message_id(message_id: str, limit: int = 14) -> str: - return truncate(message_id or "-", limit) - - -def preview_message(parts: list[dict]) -> str: - for part in parts: - if part.get("type") != "text": - continue - text = collapse_whitespace(part.get("text", "")) - if not text: - continue - prefix = "[ignored] " if part.get("ignored", False) else "" - return truncate(prefix + text) - - tool_names = [part.get("tool", "tool") for part in parts if part.get("type") == "tool"] - if tool_names: - return truncate(f"[tools: {', '.join(tool_names[:3])}]") - - for part in parts: - part_type = part.get("type", "unknown") - if part_type in {"step-start", "step-finish"}: - continue - if part_type == "tool": - tool_name = part.get("tool", "tool") - status = (part.get("state") or {}).get("status") - suffix = f" {status}" if status else "" - return f"[tool:{tool_name}{suffix}]" - return f"[{part_type}]" - - for part in parts: - part_type = part.get("type", "unknown") - return f"[{part_type}]" - - return "[no content]" - - -def extract_tool_content(part: dict) -> list[str]: - contents: list[str] = [] - tool_name = part.get("tool") - state = part.get("state") or {} - - if tool_name == "question": - questions = (state.get("input") or {}).get("questions") - if questions is not None: - content = questions if isinstance(questions, str) else stringify_json(questions) - contents.append(content) - return contents - - if tool_name in {"edit", "write"}: - if state.get("input") is not None: - input_content = state["input"] if isinstance(state["input"], str) else stringify_json(state["input"]) - contents.append(input_content) - - if state.get("status") == "completed" and state.get("output") is not None: - output = state["output"] - contents.append(output if isinstance(output, str) else stringify_json(output)) - elif state.get("status") == "error" and state.get("error") is not None: - error = state["error"] - contents.append(error if isinstance(error, str) else stringify_json(error)) - - return contents - - -def collect_message_segments(message: dict) -> tuple[list[str], int, int, list[str]]: - segments: list[str] = [] - text_segments = 0 - tool_segments = 0 - part_types: list[str] = [] - - for part in message.get("parts", []): - part_type = part.get("type", "unknown") - part_types.append(part_type) - if part_type == "text": - text = part.get("text", "") - if text: - segments.append(text) - text_segments += 1 - continue - - tool_contents = extract_tool_content(part) - segments.extend(tool_contents) - tool_segments += len(tool_contents) - - return segments, text_segments, tool_segments, part_types - - -def fallback_count_tokens(text: str) -> int: - if not text: - return 0 - return round(len(text) / 4) - - -def count_tokens_batch(texts: list[str]) -> tuple[list[int], str]: - if not texts: - return [], "anthropic" - - node_script = """ -import { countTokens } from \"@anthropic-ai/tokenizer\"; -import { readFileSync } from \"node:fs\"; - -const texts = JSON.parse(readFileSync(0, \"utf8\")); -const counts = texts.map((text) => countTokens(text || \"\")); -process.stdout.write(JSON.stringify(counts)); -""".strip() - - try: - proc = subprocess.run( - ["node", "--input-type=module", "-e", node_script], - input=stringify_json(texts), - capture_output=True, - text=True, - cwd=REPO_ROOT, - check=True, - timeout=15, - ) - counts = json.loads(proc.stdout) - if isinstance(counts, list) and len(counts) == len(texts): - return [int(count) for count in counts], "anthropic" - except (subprocess.SubprocessError, FileNotFoundError, json.JSONDecodeError, ValueError): - pass - - return [fallback_count_tokens(text) for text in texts], "approximate" - - -def get_most_recent_session(client, session_list_limit: int) -> Optional[dict]: - sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit) - return sessions[0] if sessions else None - - -def analyze_session(client, session: dict) -> dict: - session_id = session["id"] - messages = client.get_session_messages(session_id, directory=session.get("directory")) - - analyzed_messages = [] - count_inputs: list[str] = [] - for index, message in enumerate(messages, 1): - info = message.get("info", {}) - segments, text_segments, tool_segments, part_types = collect_message_segments(message) - count_inputs.append(" ".join(segments)) - analyzed_messages.append( - { - "index": index, - "message_id": info.get("id", ""), - "role": info.get("role", "unknown"), - "part_count": len(message.get("parts", [])), - "part_types": part_types, - "counted_segments": len(segments), - "text_segments": text_segments, - "tool_segments": tool_segments, - "preview": preview_message(message.get("parts", [])), - } - ) - - counts, tokenizer = count_tokens_batch(count_inputs) - total_tokens = 0 - nonzero_messages = 0 - max_tokens = 0 - - for message_data, count in zip(analyzed_messages, counts): - message_data["tokens"] = count - total_tokens += count - if count > 0: - nonzero_messages += 1 - max_tokens = max(max_tokens, count) - - return { - "session_id": session_id, - "title": session.get("title", "Unknown"), - "tokenizer": tokenizer, - "messages": analyzed_messages, - "total_messages": len(analyzed_messages), - "messages_with_tokens": nonzero_messages, - "messages_without_tokens": len(analyzed_messages) - nonzero_messages, - "total_tokens": total_tokens, - "max_message_tokens": max_tokens, - } - - -def format_token_count(count: int, colors: Colors) -> str: - c = colors - if count == 0: - return f"{c.DIM}{count:>10,}{c.RESET}" - return f"{count:>10,}" - - -def format_role(role: str, colors: Colors, width: int = 9) -> str: - c = colors - label = f"{role:<{width}}" - if role == "user": - return f"{c.CYAN}{label}{c.RESET}" - if role == "assistant": - return f"{c.GREEN}{label}{c.RESET}" - return f"{c.YELLOW}{label}{c.RESET}" - - -def format_size_indicator(count: int, max_count: int, width: int = 8) -> str: - if max_count <= 0: - return f"{'.' * width} 0%" - - pct = round((count / max_count) * 100) - if count <= 0: - filled = 0 - else: - filled = max(1, round((count / max_count) * width)) - filled = min(width, filled) - return f"{'#' * filled}{'.' * (width - filled)} {pct:>3}%" - - -def largest_messages(messages: list[dict], limit: int = 5) -> list[dict]: - return sorted(messages, key=lambda message: message.get("tokens", 0), reverse=True)[:limit] - - -def print_wide_message_table(result: dict, colors: Colors, width: int): - c = colors - messages = result["messages"] - preview_width = max(24, width - 72) - - print( - f"{c.BOLD}{'#':>3} {'Role':<9} {'Tokens':>10} {'Size':<12} {'Seg/Part':<8} {'ID':<14} Preview{c.RESET}" - ) - print("-" * width) - - for message in messages: - preview = truncate(message["preview"], preview_width) - mix = f"{message['counted_segments']}/{message['part_count']}" - print( - f"{message['index']:>3} " - f"{format_role(message['role'], c, 9)} " - f"{format_token_count(message['tokens'], c)} " - f"{format_size_indicator(message['tokens'], result['max_message_tokens']):<12} " - f"{mix:<8} " - f"{c.DIM}{short_message_id(message['message_id']):<14}{c.RESET} " - f"{preview}" - ) - - -def print_compact_message_list(result: dict, colors: Colors, width: int): - c = colors - messages = result["messages"] - meta_width = max(18, width - 6) - preview_width = max(32, width - 8) - - print(f"{c.BOLD}Messages{c.RESET}") - print("-" * width) - - for message in messages: - tokens = f"{message['tokens']:,} tokens" - size = format_size_indicator(message["tokens"], result["max_message_tokens"]) - mix = f"{message['counted_segments']}/{message['part_count']} seg/part" - meta = truncate(f"{tokens} {size} {mix}", meta_width) - preview = truncate(message["preview"], preview_width) - - print(f"{message['index']:>3} {format_role(message['role'], c, 9)} {meta}") - print(f" {c.DIM}{short_message_id(message['message_id'])}{c.RESET} {preview}") - - -def print_highlights(result: dict, colors: Colors, width: int): - c = colors - heavy_messages = [message for message in largest_messages(result["messages"]) if message.get("tokens", 0) > 0] - if not heavy_messages: - return - - print(f"\n{c.BOLD}Largest messages{c.RESET}") - print("-" * width) - for message in heavy_messages: - print( - f" #{message['index']:<3} {format_role(message['role'], c, 9)} " - f"{message['tokens']:>10,} {truncate(message['preview'], max(30, width - 33))}" - ) - - -def print_message_tokens(result: dict, colors: Colors): - c = colors - width = get_terminal_width() - print(f"{c.BOLD}{'=' * width}{c.RESET}") - print(f"{c.BOLD}SESSION MESSAGE TOKEN COUNTS{c.RESET}") - print(f"{c.BOLD}{'=' * width}{c.RESET}\n") - print(f" Session: {c.CYAN}{result['session_id']}{c.RESET}") - print(f" Title: {result['title']}") - print(f" Messages: {result['total_messages']}") - print(f" Tokenizer: {result['tokenizer']}") - print(f" Total: {result['total_tokens']:,} tokens") - print(f" Largest: {result['max_message_tokens']:,} tokens\n") - - if not result["messages"]: - print(" No messages found in this session.") - return - - if width >= 110: - print_wide_message_table(result, c, width) - else: - print_compact_message_list(result, c, width) - - print("-" * width) - print_highlights(result, c, width) - print(f"\n{c.BOLD}SESSION SUMMARY{c.RESET}") - print(f" Total message tokens: {result['total_tokens']:,}") - print(f" Messages with tokens: {result['messages_with_tokens']:,}") - print(f" Empty messages: {result['messages_without_tokens']:,}") - print(f" Largest message: {result['max_message_tokens']:,}") - - -def main() -> int: - parser = argparse.ArgumentParser( - description="Show countAllMessageTokens-style token counts for each message in an OpenCode session" - ) - parser.add_argument("--session", "-s", type=str, default=None, help="Session ID to analyze (default: most recent)") - parser.add_argument("--json", "-j", action="store_true", help="Output as JSON") - parser.add_argument("--no-color", action="store_true", help="Disable colored output") - add_api_arguments(parser) - args = parser.parse_args() - - try: - with create_client_from_args(args) as client: - if args.session is None: - session = get_most_recent_session(client, args.session_list_limit) - if session is None: - print("Error: No sessions found") - return 1 - else: - session = client.get_session(args.session) - result = analyze_session(client, session) - except APIError as err: - print(f"Error: {err}") - return 1 - - if args.json: - print(json.dumps(result, indent=2, ensure_ascii=False)) - else: - colors = NO_COLOR if args.no_color else Colors() - print_message_tokens(result, colors) - - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/opencode-session-timeline b/scripts/opencode-session-timeline deleted file mode 100755 index fff6ed19..00000000 --- a/scripts/opencode-session-timeline +++ /dev/null @@ -1,258 +0,0 @@ -#!/usr/bin/env python3 -""" -Analyze token values at each step within a single OpenCode session. -Shows cache growth over time and highlights DCP tool usage that causes cache drops. - -Queries the OpenCode SQLite database directly for fast, offline access. - -Usage: opencode-session-timeline [--session ID] [--json] [--no-color] [--db PATH] -""" - -import argparse -import json -from typing import Optional - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - - -DCP_TOOLS = { - "compress", - "prune", - "distill", - "discard", - "extract", - "context_pruning", - "squash", - "consolidate", -} - - -class Colors: - RESET = "\033[0m" - BOLD = "\033[1m" - DIM = "\033[2m" - RED = "\033[31m" - GREEN = "\033[32m" - YELLOW = "\033[33m" - CYAN = "\033[36m" - - -NO_COLOR = Colors() -for attr in dir(NO_COLOR): - if not attr.startswith("_"): - setattr(NO_COLOR, attr, "") - - -def format_duration(ms: Optional[int]) -> str: - if ms is None: - return "-" - seconds = ms / 1000 - if seconds < 60: - return f"{seconds:.1f}s" - if seconds < 3600: - minutes = int(seconds // 60) - secs = seconds % 60 - return f"{minutes}m{secs:.0f}s" - hours = int(seconds // 3600) - minutes = int((seconds % 3600) // 60) - return f"{hours}h{minutes}m" - - -def extract_step_data(parts: list[dict]) -> Optional[dict]: - step_finish = None - tools_used = [] - dcp_tools_used = [] - - for part in parts: - if part.get("type") == "step-finish" and "tokens" in part: - step_finish = part - elif part.get("type") == "tool": - tool_name = part.get("tool", "") - tools_used.append(tool_name) - if tool_name in DCP_TOOLS: - dcp_tools_used.append(tool_name) - - if step_finish is None: - return None - - tokens = step_finish.get("tokens", {}) - cache = tokens.get("cache", {}) - return { - "input": tokens.get("input", 0), - "output": tokens.get("output", 0), - "reasoning": tokens.get("reasoning", 0), - "cache_read": cache.get("read", 0), - "cache_write": cache.get("write", 0), - "cost": step_finish.get("cost", 0), - "reason": step_finish.get("reason", "unknown"), - "tools_used": tools_used, - "dcp_tools_used": dcp_tools_used, - "has_dcp": len(dcp_tools_used) > 0, - } - - -def get_most_recent_session(client, session_list_limit: int) -> Optional[dict]: - sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit) - return sessions[0] if sessions else None - - -def analyze_session(client, session: dict) -> dict: - session_id = session["id"] - messages = client.get_session_messages(session_id, directory=session.get("directory")) - title = session.get("title", "Unknown") - - steps = [] - for message in messages: - info = message.get("info", {}) - parts = message.get("parts", []) - step_data = extract_step_data(parts) - if not step_data: - continue - time_info = info.get("time", {}) - step_data["message_id"] = info.get("id", "") - step_data["created"] = time_info.get("created") - step_data["completed"] = time_info.get("completed") - steps.append(step_data) - - for idx, step in enumerate(steps): - if idx == 0: - step["cache_read_delta"] = step["cache_read"] - step["input_delta"] = step["input"] - step["time_since_prev_ms"] = None - else: - prev = steps[idx - 1] - step["cache_read_delta"] = step["cache_read"] - prev["cache_read"] - step["input_delta"] = step["input"] - prev["input"] - prev_completed = prev.get("completed") - created = step.get("created") - step["time_since_prev_ms"] = (created - prev_completed) if (prev_completed and created) else None - - created = step.get("created") - completed = step.get("completed") - step["duration_ms"] = (completed - created) if (created and completed) else None - - total_context = step["input"] + step["cache_read"] - step["cache_hit_rate"] = (step["cache_read"] / total_context * 100) if total_context > 0 else 0 - - total_input = sum(s["input"] for s in steps) - total_output = sum(s["output"] for s in steps) - total_cache_read = sum(s["cache_read"] for s in steps) - total_cost = sum(s["cost"] for s in steps) - total_duration_ms = sum((s["duration_ms"] or 0) for s in steps) - total_dcp_tools = sum(len(s["dcp_tools_used"]) for s in steps) - - total_context = total_input + total_cache_read - avg_cache_hit_rate = (total_cache_read / total_context * 100) if total_context > 0 else 0 - - return { - "session_id": session_id, - "title": title, - "steps": steps, - "total_steps": len(steps), - "total_input": total_input, - "total_output": total_output, - "total_cache_read": total_cache_read, - "total_cost": total_cost, - "total_duration_ms": total_duration_ms, - "total_dcp_tools": total_dcp_tools, - "avg_cache_hit_rate": avg_cache_hit_rate, - } - - -def print_timeline(result: dict, colors: Colors): - c = colors - print(f"{c.BOLD}{'=' * 130}{c.RESET}") - print(f"{c.BOLD}SESSION TIMELINE: Token Values at Each Step{c.RESET}") - print(f"{c.BOLD}{'=' * 130}{c.RESET}\n") - print(f" Session: {c.CYAN}{result['session_id']}{c.RESET}") - print(f" Title: {result['title']}") - print(f" Steps: {result['total_steps']}\n") - - if not result["steps"]: - print(" No steps found in this session.") - return - - print( - f"{c.BOLD}{'Step':<6} {'Cache Read':>12} {'Δ Cache':>12} {'Input':>10} {'Output':>10} {'Cache %':>9} {'Duration':>10} {'Gap':>10} {'DCP Tools':<15} {'Reason':<12}{c.RESET}" - ) - print("-" * 130) - - for idx, step in enumerate(result["steps"], 1): - cache_delta = step["cache_read_delta"] - if cache_delta > 0: - delta = f"{c.GREEN}{'+' + f'{cache_delta:,}':>11}{c.RESET}" - elif cache_delta < 0: - delta = f"{c.RED}{f'{cache_delta:,}':>12}{c.RESET}" - else: - delta = f"{c.DIM}{'0':>12}{c.RESET}" - - pct = step["cache_hit_rate"] - if pct >= 80: - pct_str = f"{c.GREEN}{pct:>8.1f}%{c.RESET}" - elif pct >= 50: - pct_str = f"{c.YELLOW}{pct:>8.1f}%{c.RESET}" - else: - pct_str = f"{c.RED}{pct:>8.1f}%{c.RESET}" - - dcp_str = f"{c.YELLOW}{', '.join(step['dcp_tools_used'])}{c.RESET}" if step["has_dcp"] else f"{c.DIM}-{c.RESET}" - row_prefix = f"{c.YELLOW}{c.BOLD}" if step["has_dcp"] else "" - row_suffix = c.RESET if step["has_dcp"] else "" - - print( - f"{row_prefix}{idx:<6}{row_suffix} {step['cache_read']:>12,} {delta} {step['input']:>10,} {step['output']:>10,} {pct_str} {format_duration(step.get('duration_ms')):>10} {format_duration(step.get('time_since_prev_ms')):>10} {dcp_str:<15} {step['reason']:<12}" - ) - - print("-" * 130) - - print(f"\n{c.BOLD}SESSION SUMMARY{c.RESET}") - print(f" Total Input Tokens: {result.get('total_input', 0):,}") - print(f" Total Output Tokens: {result.get('total_output', 0):,}") - print(f" Total Cache Read: {result.get('total_cache_read', 0):,}") - - avg_pct = result.get('avg_cache_hit_rate', 0) - if avg_pct >= 80: - avg_pct_str = f"{c.GREEN}{avg_pct:.1f}%{c.RESET}" - elif avg_pct >= 50: - avg_pct_str = f"{c.YELLOW}{avg_pct:.1f}%{c.RESET}" - else: - avg_pct_str = f"{c.RED}{avg_pct:.1f}%{c.RESET}" - - print(f" Avg Cache Hit Rate: {avg_pct_str}") - print(f" Total Cost: ${result.get('total_cost', 0):.4f}") - print(f" Total Duration: {format_duration(result.get('total_duration_ms', 0))}") - print(f" Total DCP Uses: {result.get('total_dcp_tools', 0)}") - - -def main(): - parser = argparse.ArgumentParser(description="Analyze token values at each step within an OpenCode session") - parser.add_argument("--session", "-s", type=str, default=None, help="Session ID to analyze (default: most recent)") - parser.add_argument("--json", "-j", action="store_true", help="Output as JSON") - parser.add_argument("--no-color", action="store_true", help="Disable colored output") - add_api_arguments(parser) - args = parser.parse_args() - - try: - with create_client_from_args(args) as client: - if args.session is None: - session = get_most_recent_session(client, args.session_list_limit) - if session is None: - print("Error: No sessions found") - return 1 - else: - session = client.get_session(args.session) - result = analyze_session(client, session) - except APIError as err: - print(f"Error: {err}") - return 1 - - if args.json: - print(json.dumps(result, indent=2, default=str)) - else: - colors = NO_COLOR if args.no_color else Colors() - print_timeline(result, colors) - - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/opencode-token-stats b/scripts/opencode-token-stats deleted file mode 100755 index aaf2c443..00000000 --- a/scripts/opencode-token-stats +++ /dev/null @@ -1,182 +0,0 @@ -#!/usr/bin/env python3 -""" -Analyze token usage across recent OpenCode sessions. -Usage: opencode-token-stats [--sessions N] [--json] -""" - -import json -import argparse -from datetime import datetime - -from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects - -def analyze_sessions(client, num_sessions=10, output_json=False, session_id=None, session_list_limit=5000): - # Get sessions to analyze - if session_id: - # Analyze specific session - sessions = [client.get_session(session_id)] - else: - # Get recent sessions sorted by API updated time across projects - sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit)[: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 in sessions: - session_id = session.get("id", "") - directory = session.get("directory") - 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 - messages = client.get_session_messages(session_id, directory=directory) - - for message in messages: - for part in message.get("parts", []): - if part.get("type") != "step-finish" or "tokens" not in part: - continue - 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 - - # Get session metadata (title, timestamps) - title = session.get("title", "Untitled")[:60] - created = session.get("time", {}).get("created") - - # 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") - add_api_arguments(parser) - args = parser.parse_args() - - try: - with create_client_from_args(args) as client: - analyze_sessions( - client, - num_sessions=args.sessions, - output_json=args.json, - session_id=args.session, - session_list_limit=args.session_list_limit, - ) - except APIError as err: - print(f"Error: {err}") - return 1 - - return 0 - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/opencode_api.py b/scripts/opencode_api.py deleted file mode 100644 index 8b01987f..00000000 --- a/scripts/opencode_api.py +++ /dev/null @@ -1,249 +0,0 @@ -#!/usr/bin/env python3 -"""Shared helpers for querying the OpenCode SQLite database from scripts.""" - -from __future__ import annotations - -import json -import sqlite3 -from pathlib import Path -from typing import Any - - -DEFAULT_DB_PATH = Path.home() / ".local/share/opencode/opencode.db" -DEFAULT_SESSION_LIST_LIMIT = 5000 - - -class APIError(RuntimeError): - """Script data access error (kept for backwards compatibility).""" - - def __init__(self, message: str, *, status_code: int | None = None): - super().__init__(message) - self.status_code = status_code - - -class OpencodeAPI: - """Compatibility wrapper with the old API client interface.""" - - def __init__(self, db_path: Path): - self.db_path = db_path - self.conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) - self.conn.row_factory = sqlite3.Row - - def close(self): - self.conn.close() - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - self.close() - - def health(self) -> dict[str, Any]: - self.conn.execute("SELECT 1").fetchone() - return {"status": "ok"} - - def list_projects(self) -> list[dict[str, Any]]: - rows = self.conn.execute( - "SELECT id, worktree FROM project ORDER BY time_updated DESC" - ).fetchall() - return [{"id": row["id"], "worktree": row["worktree"]} for row in rows] - - def _format_session_row(self, row: sqlite3.Row) -> dict[str, Any]: - return { - "id": row["id"], - "projectID": row["project_id"], - "parentID": row["parent_id"], - "directory": row["directory"], - "title": row["title"], - "time": {"created": row["time_created"], "updated": row["time_updated"]}, - } - - def list_sessions( - self, - *, - directory: str | None = None, - roots: bool | None = None, - start: int | None = None, - search: str | None = None, - limit: int | None = None, - ) -> list[dict[str, Any]]: - clauses: list[str] = [] - params: list[Any] = [] - - if directory: - clauses.append("directory = ?") - params.append(directory) - if roots is True: - clauses.append("parent_id IS NULL") - elif roots is False: - clauses.append("parent_id IS NOT NULL") - if search: - clauses.append("LOWER(title) LIKE ?") - params.append(f"%{search.lower()}%") - - where = f"WHERE {' AND '.join(clauses)}" if clauses else "" - query = f""" - SELECT id, project_id, parent_id, directory, title, time_created, time_updated - FROM session - {where} - ORDER BY time_updated DESC - """ - - if start is not None and start > 0: - query += " LIMIT -1 OFFSET ?" - params.append(start) - if limit is not None and limit > 0: - query = query.replace("LIMIT -1", "LIMIT ?", 1) - params[-1] = limit - params.append(start) - elif limit is not None and limit > 0: - query += " LIMIT ?" - params.append(limit) - - rows = self.conn.execute(query, params).fetchall() - return [self._format_session_row(row) for row in rows] - - def get_session(self, session_id: str, *, directory: str | None = None) -> dict[str, Any]: - params: list[Any] = [session_id] - query = """ - SELECT id, project_id, parent_id, directory, title, time_created, time_updated - FROM session - WHERE id = ? - """ - if directory: - query += " AND directory = ?" - params.append(directory) - row = self.conn.execute(query, params).fetchone() - if row is None: - raise APIError(f"Session not found: {session_id}", status_code=404) - return self._format_session_row(row) - - def get_session_messages( - self, - session_id: str, - *, - directory: str | None = None, - limit: int | None = None, - ) -> list[dict[str, Any]]: - if directory: - self.get_session(session_id, directory=directory) - - message_query = """ - SELECT id, data - FROM message - WHERE session_id = ? - ORDER BY time_created ASC - """ - message_params: list[Any] = [session_id] - if limit is not None and limit > 0: - message_query += " LIMIT ?" - message_params.append(limit) - message_rows = self.conn.execute(message_query, message_params).fetchall() - - part_rows = self.conn.execute( - """ - SELECT message_id, data - FROM part - WHERE session_id = ? - ORDER BY time_created ASC - """, - [session_id], - ).fetchall() - - parts_by_message: dict[str, list[dict[str, Any]]] = {} - for row in part_rows: - try: - payload = json.loads(row["data"]) - except (json.JSONDecodeError, TypeError): - continue - parts_by_message.setdefault(row["message_id"], []).append(payload) - - messages: list[dict[str, Any]] = [] - for row in message_rows: - try: - info = json.loads(row["data"]) - except (json.JSONDecodeError, TypeError): - info = {} - info["id"] = row["id"] - messages.append({"info": info, "parts": parts_by_message.get(row["id"], [])}) - return messages - - def get_session_message( - self, - session_id: str, - message_id: str, - *, - directory: str | None = None, - ) -> dict[str, Any]: - if directory: - self.get_session(session_id, directory=directory) - - row = self.conn.execute( - """ - SELECT data - FROM message - WHERE session_id = ? AND id = ? - """, - [session_id, message_id], - ).fetchone() - if row is None: - raise APIError(f"Message not found: {message_id}", status_code=404) - - try: - info = json.loads(row["data"]) - except (json.JSONDecodeError, TypeError): - info = {} - info["id"] = message_id - - part_rows = self.conn.execute( - """ - SELECT data - FROM part - WHERE session_id = ? AND message_id = ? - ORDER BY time_created ASC - """, - [session_id, message_id], - ).fetchall() - - parts: list[dict[str, Any]] = [] - for part_row in part_rows: - try: - parts.append(json.loads(part_row["data"])) - except (json.JSONDecodeError, TypeError): - continue - - return {"info": info, "parts": parts} - - -def add_api_arguments(parser): - parser.add_argument( - "--db", - type=str, - default=str(DEFAULT_DB_PATH), - help=f"Path to OpenCode SQLite database (default: {DEFAULT_DB_PATH})", - ) - parser.add_argument( - "--session-list-limit", - type=int, - default=DEFAULT_SESSION_LIST_LIMIT, - help="Max sessions scanned", - ) - - -def create_client_from_args(args) -> OpencodeAPI: - db_path = Path(getattr(args, "db", DEFAULT_DB_PATH)).expanduser() - if not db_path.exists(): - raise APIError(f"OpenCode database not found: {db_path}") - client = OpencodeAPI(db_path) - client.health() - return client - - -def list_sessions_across_projects( - client: OpencodeAPI, - *, - search: str | None = None, - roots: bool | None = None, - per_project_limit: int = DEFAULT_SESSION_LIST_LIMIT, -) -> list[dict[str, Any]]: - return client.list_sessions(search=search, roots=roots, limit=per_project_limit) diff --git a/tests/test-dcp-cache.sh b/tests/test-dcp-cache.sh deleted file mode 100755 index 49e7dfac..00000000 --- a/tests/test-dcp-cache.sh +++ /dev/null @@ -1,364 +0,0 @@ -#!/usr/bin/env bash -# -# DCP Token Cache Test Script -# Tests how Dynamic Context Pruning affects token caching across different providers/models -# -# Usage: -# ./test-dcp-cache.sh [OPTIONS] -# -# Options: -# --provider NAME Run test for specific provider only -# --dry-run Show what would be executed without running -# --results-dir DIR Custom results directory (default: ./results) -# --port PORT Port for server (default: 4096, enables TUI attach) -# --no-server Don't start a server, use standalone mode (no TUI attach) -# --help Show this help message -# -# To watch tests in real-time: -# Terminal 1: ./test-dcp-cache.sh --provider anthropic -# Terminal 2: opencode attach http://localhost:4096 -# - -set -euo pipefail - -# ============================================================================ -# CONFIGURATION - Modify these to change which models are tested -# ============================================================================ - -# Models to test: one per provider -# Format: ["provider-name"]="provider/model-id" -declare -A MODELS=( - ["opencode-kimi"]="opencode/kimi-k2.5-free" - ["kimi"]="kimi-for-coding/k2p5" - ["llm-proxy-cli-gemini"]="llm-proxy/cli_gemini-3-flash-high" - ["llm-proxy-ant-gemini"]="llm-proxy/ant_gemini-3-flash-high" - ["llm-proxy-opus"]="llm-proxy/claude-opus-4-5-thinking" - ["openai"]="openai/gpt-5.2-codex" - ["openrouter-haiku"]="openrouter/anthropic/claude-haiku-4.5" -) - -# Codebases to analyze (ordered from simple to complex) -# Format: "clone_command|description" -CODEBASES=( - "git clone --depth 1 https://github.com/sindresorhus/is-odd.git|is-odd: minimal npm package (~10 lines)" - "git clone --depth 1 https://github.com/chalk/chalk.git|chalk: small terminal styling utility" - "git clone --depth 1 https://github.com/tj/commander.js.git|commander: medium-complexity CLI framework" - "git clone --depth 1 https://github.com/yargs/yargs.git|yargs: medium-complex argument parser" - "cp -r ~/.config/opencode/opencode|opencode: full-featured coding assistant (local copy)" -) - -# Base prompt template - {CODEBASE_CMD} and {CODEBASE_DESC} will be replaced -PROMPT_TEMPLATE='Clone/copy {CODEBASE_DESC} to /tmp/{CODEBASE_NAME} and give me a comprehensive summary of what it does and how it works. Analyze the directory structure, key files, main functionality, and architecture. Do not use subagents.' - -# ============================================================================ -# SCRIPT LOGIC - Generally no need to modify below -# ============================================================================ - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -SCRIPTS_DIR="${SCRIPT_DIR}/../scripts" -RESULTS_DIR="${SCRIPT_DIR}/results" -DRY_RUN=false -SPECIFIC_PROVIDER="" -SERVER_PORT="4096" -USE_SERVER=true -SERVER_PID="" - -cleanup() { - if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then - log "Stopping opencode server (PID: $SERVER_PID)..." - kill "$SERVER_PID" 2>/dev/null || true - wait "$SERVER_PID" 2>/dev/null || true - fi -} -trap cleanup EXIT - -usage() { - head -20 "$0" | tail -18 | sed 's/^# \?//' - echo "" - echo "Configured models:" - for provider in "${!MODELS[@]}"; do - echo " $provider: ${MODELS[$provider]}" - done -} - -log() { - echo "[$(date '+%H:%M:%S')] $*" -} - -log_section() { - echo "" - echo "============================================================================" - echo "$*" - echo "============================================================================" -} - -# Parse arguments -while [[ $# -gt 0 ]]; do - case $1 in - --provider) - SPECIFIC_PROVIDER="$2" - shift 2 - ;; - --dry-run) - DRY_RUN=true - shift - ;; - --results-dir) - RESULTS_DIR="$2" - shift 2 - ;; - --port) - SERVER_PORT="$2" - shift 2 - ;; - --no-server) - USE_SERVER=false - shift - ;; - --help|-h) - usage - exit 0 - ;; - *) - echo "Unknown option: $1" - usage - exit 1 - ;; - esac -done - -# Create timestamp for this test run -TIMESTAMP=$(date '+%Y%m%d_%H%M%S') -RUN_DIR="${RESULTS_DIR}/${TIMESTAMP}" - -# Validate specific provider if given -if [[ -n "$SPECIFIC_PROVIDER" ]] && [[ -z "${MODELS[$SPECIFIC_PROVIDER]:-}" ]]; then - echo "Error: Unknown provider '$SPECIFIC_PROVIDER'" - echo "Available providers: ${!MODELS[*]}" - exit 1 -fi - -# Build list of providers to test -if [[ -n "$SPECIFIC_PROVIDER" ]]; then - PROVIDERS=("$SPECIFIC_PROVIDER") -else - PROVIDERS=("${!MODELS[@]}") -fi - -log_section "DCP Token Cache Test" -log "Results directory: ${RUN_DIR}" -log "Providers to test: ${PROVIDERS[*]}" -log "Codebases: ${#CODEBASES[@]}" -log "Dry run: ${DRY_RUN}" - -if [[ "$USE_SERVER" == "true" ]]; then - log "Server port: ${SERVER_PORT}" - log "" - log ">>> To watch in real-time, run in another terminal:" - log ">>> opencode attach http://localhost:${SERVER_PORT}" -else - log "Server mode: disabled (standalone runs, no TUI attach)" -fi - -if [[ "$DRY_RUN" == "true" ]]; then - log_section "DRY RUN - Commands that would be executed" -fi - -# Create results directory -if [[ "$DRY_RUN" == "false" ]]; then - mkdir -p "$RUN_DIR" - # Save test configuration (simple approach without complex jq) - { - echo "{" - echo " \"timestamp\": \"${TIMESTAMP}\"," - echo " \"providers\": [\"${PROVIDERS[*]// /\", \"}\"]," - echo " \"codebases\": ${#CODEBASES[@]}," - echo " \"server_port\": ${SERVER_PORT:-null}" - echo "}" - } > "${RUN_DIR}/config.json" -fi - -# Start server if requested -if [[ "$DRY_RUN" == "false" ]] && [[ "$USE_SERVER" == "true" ]]; then - log "" - - # Check if port is already in use - if lsof -i ":${SERVER_PORT}" &>/dev/null; then - log "Port ${SERVER_PORT} is already in use." - read -p "Kill existing process and continue? [y/N] " -n 1 -r - echo - if [[ $REPLY =~ ^[Yy]$ ]]; then - log "Killing process on port ${SERVER_PORT}..." - fuser -k "${SERVER_PORT}/tcp" 2>/dev/null || lsof -ti ":${SERVER_PORT}" | xargs -r kill -9 - sleep 1 - else - log "Aborting. Free port ${SERVER_PORT} or use --port to specify a different port." - exit 1 - fi - fi - - log "Starting opencode server on port ${SERVER_PORT}..." - opencode serve --port "$SERVER_PORT" & - SERVER_PID=$! - - # Wait for server to be ready - sleep 2 - if ! kill -0 "$SERVER_PID" 2>/dev/null; then - log "Error: Server failed to start" - exit 1 - fi - log "Server started (PID: $SERVER_PID)" -fi - -# Build base command depending on server mode -if [[ "$USE_SERVER" == "true" ]]; then - BASE_CMD="opencode run --attach http://localhost:${SERVER_PORT}" -else - BASE_CMD="opencode run" -fi - -# Run tests for each provider -for provider in "${PROVIDERS[@]}"; do - model="${MODELS[$provider]}" - provider_dir="${RUN_DIR}/${provider}" - - log_section "Testing Provider: ${provider}" - log "Model: ${model}" - - if [[ "$DRY_RUN" == "false" ]]; then - mkdir -p "$provider_dir" - fi - - SESSION_ID="" - PROMPT_NUM=0 - - for codebase_entry in "${CODEBASES[@]}"; do - PROMPT_NUM=$((PROMPT_NUM + 1)) - - # Parse codebase entry - IFS='|' read -r clone_cmd codebase_desc <<< "$codebase_entry" - codebase_name=$(echo "$clone_cmd" | grep -oE '[^/]+\.git$' | sed 's/\.git$//' || echo "$clone_cmd" | awk '{print $NF}') - - # Build the prompt - prompt="${PROMPT_TEMPLATE}" - prompt="${prompt//\{CODEBASE_CMD\}/$clone_cmd}" - prompt="${prompt//\{CODEBASE_DESC\}/$codebase_desc}" - prompt="${prompt//\{CODEBASE_NAME\}/$codebase_name}" - - log "" - log "Prompt ${PROMPT_NUM}/${#CODEBASES[@]}: ${codebase_desc}" - - # Build opencode command (for display only, actual execution below) - if [[ -z "$SESSION_ID" ]]; then - # First prompt - create new session - display_cmd="${BASE_CMD} -m '${model}' --title 'DCP Test: ${provider}' ''" - else - # Subsequent prompts - continue session - display_cmd="${BASE_CMD} -m '${model}' --session '${SESSION_ID}' ''" - fi - - if [[ "$DRY_RUN" == "true" ]]; then - echo " $ $display_cmd" - # Simulate session ID for dry run - if [[ -z "$SESSION_ID" ]]; then - SESSION_ID="dry-run-session-id" - fi - else - log "Executing: $display_cmd" - - # Run opencode and capture output - output_file="${provider_dir}/prompt_${PROMPT_NUM}_output.txt" - json_file="${provider_dir}/prompt_${PROMPT_NUM}_events.json" - - if [[ -z "$SESSION_ID" ]]; then - # First run - use --format json to capture session ID from events - log "Using JSON format to capture session ID..." - if [[ "$USE_SERVER" == "true" ]]; then - opencode run --attach "http://localhost:${SERVER_PORT}" \ - -m "${model}" \ - --title "DCP Test: ${provider}" \ - --format json \ - "${prompt}" 2>&1 | tee "$json_file" - else - opencode run \ - -m "${model}" \ - --title "DCP Test: ${provider}" \ - --format json \ - "${prompt}" 2>&1 | tee "$json_file" - fi - - # Extract session ID from the first JSON event - SESSION_ID=$(head -1 "$json_file" | jq -r '.sessionID // empty' 2>/dev/null || echo "") - - if [[ -z "$SESSION_ID" ]]; then - log "Warning: Could not extract session ID from JSON output" - # Fallback to opencode-find-session - log "Falling back to session search..." - SESSION_ID=$("${SCRIPTS_DIR}/opencode-find-session" "DCP Test: ${provider}" 2>/dev/null | head -1 || echo "") - fi - - if [[ -z "$SESSION_ID" ]]; then - log "Error: Could not find session ID, cannot continue session" - log "Will create new sessions for each prompt (cache test will be less meaningful)" - fi - - log "Session ID: ${SESSION_ID:-unknown}" - echo "$SESSION_ID" > "${provider_dir}/session_id.txt" - else - # Subsequent prompts - continue session with normal output - if [[ "$USE_SERVER" == "true" ]]; then - opencode run --attach "http://localhost:${SERVER_PORT}" \ - -m "${model}" \ - --session "${SESSION_ID}" \ - "${prompt}" 2>&1 | tee "$output_file" - else - opencode run \ - -m "${model}" \ - --session "${SESSION_ID}" \ - "${prompt}" 2>&1 | tee "$output_file" - fi - fi - - log "Output saved to: ${output_file:-$json_file}" - fi - done - - # Collect analysis after all prompts for this provider - if [[ "$DRY_RUN" == "false" ]] && [[ -n "$SESSION_ID" ]]; then - log "" - log "Collecting cache analysis for session ${SESSION_ID}..." - - # Session timeline - "${SCRIPTS_DIR}/opencode-session-timeline" --session "$SESSION_ID" --no-color > "${provider_dir}/session_timeline.txt" 2>&1 || true - "${SCRIPTS_DIR}/opencode-session-timeline" --session "$SESSION_ID" --json > "${provider_dir}/session_timeline.json" 2>&1 || true - - # Token stats - "${SCRIPTS_DIR}/opencode-token-stats" --session "$SESSION_ID" > "${provider_dir}/token_stats.txt" 2>&1 || true - "${SCRIPTS_DIR}/opencode-token-stats" --session "$SESSION_ID" --json > "${provider_dir}/token_stats.json" 2>&1 || true - - # DCP stats - "${SCRIPTS_DIR}/opencode-dcp-stats" --session "$SESSION_ID" > "${provider_dir}/dcp_stats.txt" 2>&1 || true - "${SCRIPTS_DIR}/opencode-dcp-stats" --session "$SESSION_ID" --json > "${provider_dir}/dcp_stats.json" 2>&1 || true - - log "Analysis saved to: ${provider_dir}/" - elif [[ "$DRY_RUN" == "true" ]]; then - echo "" - echo " # After session completes:" - echo " $ ${SCRIPTS_DIR}/opencode-session-timeline --session \$SESSION_ID > ${provider_dir}/session_timeline.txt" - echo " $ ${SCRIPTS_DIR}/opencode-token-stats --session \$SESSION_ID > ${provider_dir}/token_stats.txt" - echo " $ ${SCRIPTS_DIR}/opencode-dcp-stats --session \$SESSION_ID > ${provider_dir}/dcp_stats.txt" - fi -done - -log_section "Test Complete" -if [[ "$DRY_RUN" == "false" ]]; then - log "Results saved to: ${RUN_DIR}" - log "" - log "To view results:" - echo " ls -la ${RUN_DIR}/" - for provider in "${PROVIDERS[@]}"; do - echo " cat ${RUN_DIR}/${provider}/session_timeline.txt" - done -else - log "Dry run complete. Run without --dry-run to execute tests." -fi