opencode-runtime.py

#!/usr/bin/env python3
"""
OpenCode Runtime Analyzer

SPDX-License-Identifier: MIT

A utility for reconstructing active agent and tool runtime from
OpenCode session data, including support for Superpowers Subagent-Driven 
Development (SDD) programme progress reporting.

opencode-runtime.py

Estimate active OpenCode runtime for a codebase from OpenCode's SQLite database.

Key metrics:
  * Active elapsed: union of all persisted agent/tool intervals. Concurrent work
    is counted once and long periods with no recorded activity are excluded.
  * Aggregate agent time: sum of agent intervals. Concurrent agents count
    separately.
  * Tool time: sum of persisted tool-call intervals.
  * Programme progress: counts Task headings in docs/superpowers/plans and
    completed task codes in .superpowers/sdd/**/progress.md.

The database is opened read-only. A live WAL database is supported by making a
temporary SQLite backup first unless --no-backup is supplied.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import sqlite3
import sys
import tempfile
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Iterator, Optional


@dataclass(frozen=True, order=True)
class Interval:
    start_ms: int
    end_ms: int
    category: str
    session_id: str
    label: str = ""

    @property
    def seconds(self) -> float:
        return max(0, self.end_ms - self.start_ms) / 1000.0


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Report active OpenCode agent/tool runtime for a codebase."
    )
    p.add_argument("root", type=Path, help="Codebase/worktree root")
    p.add_argument(
        "--db",
        type=Path,
        default=Path.home() / ".local/share/opencode/opencode.db",
        help="OpenCode SQLite DB (default: ~/.local/share/opencode/opencode.db)",
    )
    p.add_argument(
        "--session",
        action="append",
        default=[],
        help="Restrict to this session ID; repeat for multiple sessions",
    )
    p.add_argument(
        "--idle-threshold",
        type=float,
        default=300.0,
        metavar="SECONDS",
        help=(
            "Maximum uninstrumented gap retained inside an agent turn "
            "(default: 300). Persisted tool/part intervals are never truncated."
        ),
    )
    p.add_argument(
        "--no-backup",
        action="store_true",
        help="Read the live DB directly instead of making a temporary backup",
    )
    p.add_argument("--json", action="store_true", help="Emit JSON")
    p.add_argument("--verbose", action="store_true")
    return p.parse_args()


def norm_path(p: Path | str) -> str:
    return os.path.realpath(os.path.expanduser(str(p))).rstrip(os.sep)


def json_obj(raw: Any) -> dict[str, Any]:
    if isinstance(raw, dict):
        return raw
    if raw is None:
        return {}
    try:
        value = json.loads(raw)
        return value if isinstance(value, dict) else {}
    except (TypeError, ValueError, json.JSONDecodeError):
        return {}


def nested(obj: dict[str, Any], *keys: str) -> Any:
    cur: Any = obj
    for key in keys:
        if not isinstance(cur, dict):
            return None
        cur = cur.get(key)
    return cur


def as_ms(value: Any) -> Optional[int]:
    if value is None or isinstance(value, bool):
        return None
    try:
        n = float(value)
    except (TypeError, ValueError):
        return None
    # Current OpenCode timestamps are Unix milliseconds. Accept seconds too.
    if 1_000_000_000 <= n < 10_000_000_000:
        n *= 1000
    if n <= 0:
        return None
    return int(n)


def interval_from_times(
    times: Any,
    category: str,
    session_id: str,
    label: str = "",
) -> Optional[Interval]:
    if not isinstance(times, dict):
        return None
    start = as_ms(times.get("start") or times.get("created"))
    end = as_ms(
        times.get("end")
        or times.get("completed")
        or times.get("finished")
        or times.get("updated")
    )
    if start is None or end is None or end < start:
        return None
    return Interval(start, end, category, session_id, label)


def table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
    return {row[1] for row in conn.execute(f'PRAGMA table_info("{table}")')}


def existing_tables(conn: sqlite3.Connection) -> set[str]:
    return {
        row[0]
        for row in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )
    }


def make_db_copy(db: Path, no_backup: bool) -> tuple[Path, Optional[tempfile.TemporaryDirectory]]:
    if no_backup:
        return db, None

    td = tempfile.TemporaryDirectory(prefix="opencode-runtime-")
    dest = Path(td.name) / "opencode.db"
    try:
        src_uri = f"file:{db}?mode=ro"
        with sqlite3.connect(src_uri, uri=True, timeout=30) as src:
            with sqlite3.connect(dest) as out:
                src.backup(out)
        return dest, td
    except sqlite3.Error:
        td.cleanup()
        raise


def select_sessions(
    conn: sqlite3.Connection,
    root: str,
    requested: list[str],
) -> list[dict[str, Any]]:
    if "session" not in existing_tables(conn):
        raise RuntimeError("Database has no 'session' table")

    cols = table_columns(conn, "session")
    needed = ["id"]
    for c in ("parent_id", "directory", "title", "time_created", "time_updated"):
        if c in cols:
            needed.append(c)

    rows = [
        dict(zip(needed, row))
        for row in conn.execute(
            "SELECT " + ", ".join(f'"{c}"' for c in needed) + " FROM session"
        )
    ]
    by_id = {r["id"]: r for r in rows}

    if requested:
        missing = [sid for sid in requested if sid not in by_id]
        if missing:
            raise RuntimeError("Unknown session ID(s): " + ", ".join(missing))
        selected_ids = set(requested)
    else:
        selected_ids: set[str] = set()
        prefix = root + os.sep
        for r in rows:
            directory = r.get("directory")
            if not directory:
                continue
            d = norm_path(directory)
            # Include the root, nested worktrees, and sessions whose stored
            # directory is a parent of the supplied nested worktree.
            if d == root or d.startswith(prefix) or root.startswith(d + os.sep):
                selected_ids.add(r["id"])

    # Include all descendant/subagent sessions.
    changed = True
    while changed:
        changed = False
        for r in rows:
            if r.get("parent_id") in selected_ids and r["id"] not in selected_ids:
                selected_ids.add(r["id"])
                changed = True

    return [by_id[sid] for sid in selected_ids]


def query_in_chunks(
    conn: sqlite3.Connection,
    sql_prefix: str,
    ids: list[str],
    chunk_size: int = 800,
) -> Iterator[sqlite3.Row]:
    for i in range(0, len(ids), chunk_size):
        chunk = ids[i : i + chunk_size]
        placeholders = ",".join("?" for _ in chunk)
        yield from conn.execute(sql_prefix.format(placeholders=placeholders), chunk)


def collect_intervals(
    conn: sqlite3.Connection,
    sessions: list[dict[str, Any]],
    idle_threshold_s: float,
    verbose: bool,
) -> tuple[list[Interval], Counter[str], dict[str, dict[str, Any]]]:
    session_ids = [s["id"] for s in sessions]
    message_meta: dict[str, dict[str, Any]] = {}
    intervals: list[Interval] = []
    models: Counter[str] = Counter()

    tables = existing_tables(conn)

    if "message" in tables:
        mcols = table_columns(conn, "message")
        select = ["id", "session_id", "data"]
        for c in ("time_created", "time_updated"):
            if c in mcols:
                select.append(c)
        sql = (
            "SELECT " + ", ".join(f'"{c}"' for c in select) +
            " FROM message WHERE session_id IN ({placeholders})"
        )
        for row in query_in_chunks(conn, sql, session_ids):
            rec = dict(zip(select, row))
            data = json_obj(rec.get("data"))
            sid = rec["session_id"]
            message_meta[rec["id"]] = {"session_id": sid, "data": data, **rec}

            role = data.get("role")
            provider = data.get("providerID") or nested(data, "model", "providerID")
            model = data.get("modelID") or nested(data, "model", "modelID")
            if provider or model:
                models[f"{provider or '?'} / {model or '?'}"] += 1

            if role == "assistant":
                times = data.get("time", {})
                start = as_ms(
                    nested(data, "time", "created")
                    or rec.get("time_created")
                )
                end = as_ms(
                    nested(data, "time", "completed")
                    or nested(data, "time", "finished")
                    or rec.get("time_updated")
                )
                if start and end and end >= start:
                    # A fallback agent-turn interval. It is capped below when
                    # the only evidence is a large silent span.
                    max_ms = int(idle_threshold_s * 1000)
                    if end - start > max_ms:
                        end = start + max_ms
                    intervals.append(
                        Interval(start, end, "agent-turn", sid, model or "")
                    )

    if "part" in tables:
        pcols = table_columns(conn, "part")
        select = ["id", "message_id", "session_id", "data"]
        for c in ("time_created", "time_updated"):
            if c in pcols:
                select.append(c)
        sql = (
            "SELECT " + ", ".join(f'"{c}"' for c in select) +
            " FROM part WHERE session_id IN ({placeholders})"
        )
        for row in query_in_chunks(conn, sql, session_ids):
            rec = dict(zip(select, row))
            data = json_obj(rec.get("data"))
            sid = rec["session_id"]
            ptype = str(data.get("type") or "unknown")

            if ptype == "tool":
                state = data.get("state") if isinstance(data.get("state"), dict) else {}
                label = str(data.get("tool") or state.get("tool") or "tool")
                iv = interval_from_times(state.get("time"), "tool", sid, label)
                if iv:
                    intervals.append(iv)
                    continue

            category = {
                "reasoning": "thinking",
                "text": "generation",
                "step-start": "agent-other",
                "step-finish": "agent-other",
            }.get(ptype)

            if category:
                iv = interval_from_times(data.get("time"), category, sid, ptype)
                if iv:
                    intervals.append(iv)
                    continue

            # Some versions put timestamps at the part row level.
            start = as_ms(rec.get("time_created"))
            end = as_ms(rec.get("time_updated"))
            if start and end and end >= start and ptype in {"reasoning", "text"}:
                intervals.append(
                    Interval(
                        start,
                        end,
                        "thinking" if ptype == "reasoning" else "generation",
                        sid,
                        ptype,
                    )
                )

    if verbose:
        print(
            f"[debug] loaded {len(message_meta)} messages and "
            f"{len(intervals)} intervals",
            file=sys.stderr,
        )
    return intervals, models, message_meta


def merge_intervals(items: Iterable[Interval]) -> list[tuple[int, int]]:
    pairs = sorted((i.start_ms, i.end_ms) for i in items if i.end_ms >= i.start_ms)
    merged: list[list[int]] = []
    for start, end in pairs:
        if not merged or start > merged[-1][1]:
            merged.append([start, end])
        else:
            merged[-1][1] = max(merged[-1][1], end)
    return [(a, b) for a, b in merged]


def duration_of_merged(items: Iterable[Interval]) -> float:
    return sum((b - a) / 1000 for a, b in merge_intervals(items))


def fmt_duration(seconds: float) -> str:
    seconds = max(0, int(round(seconds)))
    days, rem = divmod(seconds, 86400)
    hours, rem = divmod(rem, 3600)
    minutes, secs = divmod(rem, 60)
    if days:
        return f"{days}d {hours:02d}h {minutes:02d}m {secs:02d}s"
    if hours:
        return f"{hours}h {minutes:02d}m {secs:02d}s"
    if minutes:
        return f"{minutes}m {secs:02d}s"
    return f"{secs}s"


TASK_HEADING = re.compile(
    r"^#{1,6}\s+Task\s+(\d+)(?:\s*[:.\-]|\s|$)", re.IGNORECASE
)
COMPLETE_CODE = re.compile(
    r"\b([AFILNP]\d+)\b.*?\b(?:complete|completed|done|passed)\b",
    re.IGNORECASE,
)
ACTIVE_CODE = re.compile(
    r"\b([AFILNP]\d+)\b.*?\b(?:active|current|in[ -]?progress|running)\b",
    re.IGNORECASE,
)


def plan_progress(root: Path) -> dict[str, Any]:
    plans_dir = root / "docs/superpowers/plans"
    plans: list[dict[str, Any]] = []

    prefix_rules = [
        ("foundation-configuration", "F"),
        ("ingestion-runs-snapshots", "I"),
        ("physical-sweep-execution", "P"),
        ("loans-interest-allocation", "L"),
        ("accounting-reconciliation", "A"),
        ("notional-reporting-operations", "N"),
    ]

    if plans_dir.is_dir():
        for f in sorted(plans_dir.glob("*-plan.md")):
            if "master-plan" in f.name:
                continue
            prefix = "?"
            for needle, candidate in prefix_rules:
                if needle in f.name:
                    prefix = candidate
                    break
            count = 0
            try:
                with f.open(encoding="utf-8", errors="replace") as fh:
                    count = sum(1 for line in fh if TASK_HEADING.match(line))
            except OSError:
                pass
            plans.append({"prefix": prefix, "tasks": count, "file": f.name})

    completed: set[str] = set()
    active: set[str] = set()
    sdd_dir = root / ".superpowers/sdd"
    if sdd_dir.exists():
        for f in sdd_dir.rglob("progress.md"):
            try:
                text = f.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            for line in text.splitlines():
                completed.update(m.group(1).upper() for m in COMPLETE_CODE.finditer(line))
                active.update(m.group(1).upper() for m in ACTIVE_CODE.finditer(line))

    total = sum(p["tasks"] for p in plans)
    return {
        "plans": plans,
        "total": total,
        "completed_codes": sorted(completed),
        "completed": len(completed),
        "active_codes": sorted(active - completed),
    }


def timestamp(ms: int) -> str:
    return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).isoformat()


def build_report(args: argparse.Namespace) -> dict[str, Any]:
    root = norm_path(args.root)
    db = args.db.expanduser().resolve()
    if not Path(root).is_dir():
        raise RuntimeError(f"Codebase root does not exist: {root}")
    if not db.is_file():
        raise RuntimeError(f"OpenCode database does not exist: {db}")

    copy_path, temporary = make_db_copy(db, args.no_backup)
    try:
        conn = sqlite3.connect(f"file:{copy_path}?mode=ro", uri=True)
        try:
            sessions = select_sessions(conn, root, args.session)
            if not sessions:
                raise RuntimeError(
                    "No OpenCode sessions matched this root. Try --session ses_..."
                )
            intervals, models, _ = collect_intervals(
                conn, sessions, args.idle_threshold, args.verbose
            )
        finally:
            conn.close()
    finally:
        if temporary is not None:
            temporary.cleanup()

    if not intervals:
        raise RuntimeError(
            "Sessions were found, but no persisted timing intervals were found."
        )

    first = min(i.start_ms for i in intervals)
    last = max(i.end_ms for i in intervals)
    wall = (last - first) / 1000
    active_union = duration_of_merged(intervals)

    categories: dict[str, float] = {}
    for category in sorted({i.category for i in intervals}):
        categories[category] = sum(
            i.seconds for i in intervals if i.category == category
        )

    tools = Counter()
    for i in intervals:
        if i.category == "tool":
            tools[i.label or "tool"] += i.seconds

    per_session: list[dict[str, Any]] = []
    for s in sessions:
        these = [i for i in intervals if i.session_id == s["id"]]
        if not these:
            continue
        per_session.append(
            {
                "id": s["id"],
                "parent_id": s.get("parent_id"),
                "title": s.get("title", ""),
                "directory": s.get("directory", ""),
                "active_elapsed_seconds": duration_of_merged(these),
                "aggregate_seconds": sum(i.seconds for i in these),
            }
        )

    aggregate = sum(i.seconds for i in intervals)
    return {
        "root": root,
        "database": str(db),
        "sessions": len(per_session),
        "subagents": sum(1 for s in per_session if s.get("parent_id")),
        "first_activity_utc": timestamp(first),
        "last_activity_utc": timestamp(last),
        "wall_clock_seconds": wall,
        "active_elapsed_seconds": active_union,
        "idle_excluded_seconds": max(0, wall - active_union),
        "aggregate_recorded_seconds": aggregate,
        "utilisation_percent": (100 * active_union / wall) if wall else 100.0,
        "categories": categories,
        "tools": dict(tools.most_common()),
        "models": dict(models.most_common()),
        "programme": plan_progress(Path(root)),
        "per_session": sorted(
            per_session,
            key=lambda x: x["active_elapsed_seconds"],
            reverse=True,
        ),
        "notes": [
            "Active elapsed merges concurrent agent/tool intervals, so parallel work is counted once.",
            "Aggregate recorded time sums intervals, so parallel agents count separately.",
            "Tool time may overlap agent-turn time and is therefore not additive with it.",
            "Agent-turn spans longer than --idle-threshold are capped when finer part timing is unavailable.",
        ],
    }


def print_text(r: dict[str, Any]) -> None:
    width = 66
    line = "═" * width
    thin = "─" * width

    print(line)
    print("OpenCode Runtime Report")
    print()
    print(f"Repository: {r['root']}")
    print(f"Sessions:   {r['sessions']} ({r['subagents']} subagent sessions)")
    print(f"From:       {r['first_activity_utc']}")
    print(f"To:         {r['last_activity_utc']}")
    print()
    print(f"{'Wall clock':28} {fmt_duration(r['wall_clock_seconds']):>20}")
    print(f"{'Idle/off time excluded':28} {fmt_duration(r['idle_excluded_seconds']):>20}")
    print(thin)
    print(f"{'Active elapsed (union)':28} {fmt_duration(r['active_elapsed_seconds']):>20}")
    print(f"{'Aggregate recorded time':28} {fmt_duration(r['aggregate_recorded_seconds']):>20}")
    print(f"{'Utilisation':28} {r['utilisation_percent']:>19.1f}%")

    print()
    print("Recorded Runtime by Category")
    for name, seconds in sorted(
        r["categories"].items(), key=lambda kv: kv[1], reverse=True
    ):
        print(f"{name:28} {fmt_duration(seconds):>20}")

    if r["tools"]:
        print()
        print("Tool Runtime")
        for name, seconds in r["tools"].items():
            print(f"{name[:28]:28} {fmt_duration(seconds):>20}")

    if r["models"]:
        print()
        print("Model Calls / Assistant Messages")
        for name, count in r["models"].items():
            print(f"{name[:48]:48} {count:>8}")

    prog = r["programme"]
    if prog["plans"]:
        print()
        print("Programme Progress")
        for p in prog["plans"]:
            prefix = p["prefix"]
            task_range = (
                f"{prefix}1–{prefix}{p['tasks']}" if p["tasks"] else f"{prefix}?"
            )
            print(f"{task_range:10} {p['tasks']:>3} tasks  {p['file']}")
        if prog["total"]:
            pct = 100 * prog["completed"] / prog["total"]
            print(thin)
            print(
                f"{'Completed from progress.md':28} "
                f"{prog['completed']} / {prog['total']} ({pct:.1f}%)"
            )
            if prog["active_codes"]:
                print(f"{'Active':28} {', '.join(prog['active_codes'])}")

    print()
    print("Longest Sessions")
    for s in r["per_session"][:15]:
        kind = "subagent" if s.get("parent_id") else "parent"
        title = (s.get("title") or "").replace("\n", " ")[:34]
        print(
            f"{fmt_duration(s['active_elapsed_seconds']):>12}  "
            f"{kind:8}  {s['id']}  {title}"
        )

    print()
    for note in r["notes"]:
        print(f"* {note}")
    print(line)


def main() -> int:
    args = parse_args()
    try:
        report = build_report(args)
        if args.json:
            print(json.dumps(report, indent=2))
        else:
            print_text(report)
        return 0
    except (RuntimeError, sqlite3.Error, OSError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())