refactor: migrate scripts to OpenCode session API

Author: Tarquinen

.gitignore

@@ -28,6 +28,11 @@ Thumbs.db
 # OpenCode
 .opencode/
 
+# Python cache
+__pycache__/
+*.py[cod]
+*$py.class
+
 # Generated prompt files (from scripts/generate-prompts.ts)
 lib/prompts/**/*.generated.ts
 
@@ -40,4 +45,4 @@ test-update.ts
 docs/
 SCHEMA_NOTES.md
 
-repomix-output.xml
+repomix-output.xml

scripts/opencode-dcp-stats

@@ -8,11 +8,12 @@ Usage: opencode-dcp-stats [--sessions N] [--min-messages M] [--json] [--verbose]
 
 import json
 import argparse
-from pathlib import Path
 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",
@@ -24,42 +25,9 @@ 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(storage: Path, session_id: str) -> list[dict]:
-    """Get all messages for a session, sorted by creation order."""
-    message_dir = storage / "message" / session_id
-    if not message_dir.exists():
-        return []
-    
-    messages = []
-    for msg_file in message_dir.glob("*.json"):
-        try:
-            msg = json.loads(msg_file.read_text())
-            msg["_file"] = msg_file
-            msg["_id"] = msg_file.stem
-            messages.append(msg)
-        except (json.JSONDecodeError, IOError):
-            pass
-    
-    return sorted(messages, key=lambda m: m.get("_id", ""))
-
-
-def get_message_parts(storage: Path, message_id: str) -> list[dict]:
-    """Get all parts for a message, sorted by creation order."""
-    parts_dir = storage / "part" / message_id
-    if not parts_dir.exists():
-        return []
-    
-    parts = []
-    for part_file in parts_dir.glob("*.json"):
-        try:
-            part = json.loads(part_file.read_text())
-            part["_file"] = part_file
-            part["_id"] = part_file.stem
-            parts.append(part)
-        except (json.JSONDecodeError, IOError):
-            pass
-    
-    return sorted(parts, key=lambda p: p.get("_id", ""))
+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:
@@ -83,22 +51,20 @@ def is_ignored_message(message: dict, parts: list[dict]) -> bool:
     return True
 
 
-def count_real_user_messages(storage: Path, session_id: str) -> int:
+def count_real_user_messages(messages: list[dict]) -> int:
     """Count user messages that are not ignored (real user interactions)."""
-    messages = get_session_messages(storage, session_id)
     count = 0
-    
+
     for msg in messages:
+        info = msg.get("info", {})
+        parts = msg.get("parts", [])
         # Only count user role messages
-        if msg.get("role") != "user":
+        if info.get("role") != "user":
             continue
-        
-        msg_id = msg.get("_id", "")
-        parts = get_message_parts(storage, msg_id)
-        
+
         if not is_ignored_message(msg, parts):
             count += 1
-    
+
     return count
 
 
@@ -136,10 +102,9 @@ def calc_cache_hit_rate(tokens: dict) -> float:
     return (cache_read / total_context) * 100
 
 
-def analyze_session(storage: Path, session_id: str) -> dict:
+def analyze_session(messages: list[dict], session_id: str) -> dict:
     """Analyze DCP impact for a single session."""
-    messages = get_session_messages(storage, session_id)
-    
+
     result = {
         "session_id": session_id,
         "dcp_events": [],
@@ -164,14 +129,15 @@ def analyze_session(storage: Path, session_id: str) -> dict:
     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_id = msg.get("_id", "")
-        parts = get_message_parts(storage, msg_id)
-        
+        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", {})
@@ -239,27 +205,24 @@ def analyze_session(storage: Path, session_id: str) -> dict:
     return result
 
 
-def analyze_sessions(num_sessions: int = 20, min_messages: int = 5, output_json: bool = False, verbose: bool = False, session_id: str = None):
+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."""
-    storage = Path.home() / ".local/share/opencode/storage"
-    message_dir = storage / "message"
-    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]
+        sessions = [client.get_session(session_id)]
     else:
-        sessions = sorted(message_dir.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True)[:num_sessions]
-    
+        sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit)[:num_sessions]
+
     all_results = []
     grand_totals = {
         "sessions_analyzed": 0,
@@ -283,44 +246,34 @@ def analyze_sessions(num_sessions: int = 20, min_messages: int = 5, output_json:
         "hit_rates_by_distance": defaultdict(list)
     }
 
-    for session_path in sessions:
-        session_id = session_path.name
-        
+    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(storage, session_id)
+        real_user_messages = count_real_user_messages(messages)
         if real_user_messages < min_messages:
             grand_totals["sessions_skipped_short"] += 1
             continue
-        
-        result = analyze_session(storage, session_id)
+
+        result = analyze_session(messages, session_id)
         result["user_messages"] = real_user_messages
-        
-        # Get session metadata
-        title = "Unknown"
-        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")[:50]
-                except (json.JSONDecodeError, IOError):
-                    pass
-                break
-        
-        result["title"] = title
-        
+
+        # 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)
@@ -474,16 +427,26 @@ def main():
                         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()
-    
-    analyze_sessions(
-        num_sessions=args.sessions,
-        min_messages=args.min_messages,
-        output_json=args.json,
-        verbose=args.verbose,
-        session_id=args.session
-    )
+
+    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__":
-    main()
+    raise SystemExit(main())

scripts/opencode-find-session

@@ -1,53 +1,35 @@
 #!/usr/bin/env python3
 """
-Find OpenCode session IDs by title search.
+Find OpenCode session IDs by title search using the OpenCode API.
 Returns matching session IDs ordered by last usage time.
 
 Usage: opencode-find-session <search-term> [--exact] [--json]
 """
 
 import json
 import argparse
-from pathlib import Path
 from datetime import datetime
 
+from opencode_api import APIError, add_api_arguments, create_client_from_args, list_sessions_across_projects
 
-def get_all_sessions(storage: Path) -> list[dict]:
-    """Get all sessions with their metadata."""
-    session_dir = storage / "session"
-    message_dir = storage / "message"
-    
-    if not session_dir.exists():
-        return []
-    
+
+def get_all_sessions(client, session_list_limit: int) -> list[dict]:
+    """Get all sessions with normalized metadata from API."""
+    api_sessions = list_sessions_across_projects(client, per_project_limit=session_list_limit)
     sessions = []
-    
-    for app_dir in session_dir.iterdir():
-        if not app_dir.is_dir():
-            continue
-        
-        for session_file in app_dir.glob("*.json"):
-            try:
-                session = json.loads(session_file.read_text())
-                session_id = session_file.stem
-                
-                # Get last modified time from message directory
-                msg_path = message_dir / session_id
-                if msg_path.exists():
-                    mtime = msg_path.stat().st_mtime
-                else:
-                    mtime = session_file.stat().st_mtime
-                
-                sessions.append({
-                    "id": session_id,
-                    "title": session.get("title", "Untitled"),
-                    "created_at": session.get("createdAt"),
-                    "last_used": mtime,
-                    "last_used_iso": datetime.fromtimestamp(mtime).isoformat()
-                })
-            except (json.JSONDecodeError, IOError):
-                pass
-    
+    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
 
 
@@ -101,6 +83,7 @@ def main():
     )
     parser.add_argument(
         "search_term",
+        nargs="?",
         type=str,
         help="Text to search for in session titles"
     )
@@ -119,16 +102,19 @@ def main():
         action="store_true",
         help="Show all sessions (ignore search term)"
     )
+    add_api_arguments(parser)
     args = parser.parse_args()
-    
-    storage = Path.home() / ".local/share/opencode/storage"
-    
-    if not storage.exists():
-        print("Error: OpenCode storage not found at", storage)
+
+    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
-    
-    sessions = get_all_sessions(storage)
-    
+
     if args.all:
         results = sorted(sessions, key=lambda s: s["last_used"], reverse=True)
     else: