6krrt/scripts/export_progress_fixture.py

#!/usr/bin/env python3
"""Export labelled calibration sessions from opencode SQLite DB into fixture JSON.

Usage:
    python scripts/export_progress_fixture.py

Output: tests/fixtures/progress/fixture.json

Fixture format: array of {session_id, label, root_id, title, agent, file_lines, calls}

Content scrubbing: sha1 everything except filePath/path/offset/limit/command/pattern.
These keys are the detector's target and must pass through verbatim.
"""

import hashlib
import json
import sqlite3
import subprocess
import sys
from pathlib import Path
from typing import Any

DB_PATH = Path.home() / ".local" / "share" / "opencode" / "opencode.db"
OUTPUT_PATH = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "progress" / "fixture.json"

GIT_REPO = "/home/alee/Sources/6krrt"
WORKTREE_PREFIX = "/home/alee/Sources/6krrt-cockpit-quick-wins/"
MAIN_REPO_PREFIX = "/home/alee/Sources/6krrt/"
WORKTREE_COMMIT = "827c408"

# Regex to find likely file paths in bash command text
import re

_PATH_RE = re.compile(r"/home/alee/Sources/6krrt(?:-cockpit-quick-wins)?/[\w./-]+\.(?:py|js|html|css|json|yaml|yml|md|toml|sh|sql|mjs|ts|tsx)")

LABELS: dict[str, str] = {
    "ses_f24cc0258ffepbJPoOzoiLZi9s": "must_flag",
    "ses_f24fcf651ffeOm9O4E14QSZeJ4": "must_flag",
    "ses_f24890640ffewKnIPqjjAfFvOk": "must_flag",
    "ses_f245607eaffeGxPH8F1tvvdvkG": "must_flag",
    "ses_f237e7e31ffeOzz4kUKu4CWzMu": "must_flag",
    "ses_f235d755bffeo0uWtD1g2HwQUg": "must_flag",
    "ses_f2506ac70ffebLUxBsq0aauOn9": "must_flag",
    "ses_f2380dd4cffeVwcZn3Kv3nQN7v": "must_flag",
    "ses_f246e699fffeCHUPTLr5ajBhWC": "must_not_flag",
    "ses_f245f78f8ffeGMn3PgArpOwhFy": "must_not_flag",
    "ses_f243ea08effemMwXdMRE1gpnhK": "must_not_flag",
    "ses_f242e5e28ffehF97ZxxdbTeylQ": "must_not_flag",
    "ses_f236bbdaaffezwnvH10X2rgrxc": "must_not_flag",
    "ses_f236bc7baffeG5LQsr2K1c46xO": "must_not_flag",
    "ses_f2e81929fffeaVbWui0X9QjmRQ": "must_not_flag",
}

SAFE_KEYS = {"filePath", "path", "offset", "limit", "command", "pattern", "output_mode", "include", "tmux_command"}

# ── line count cache ────────────────────────────────────────────────────────

_LEN_CACHE: dict[str, int | None] = {}


def _count_lines_worktree(relpath: str) -> int | None:
    """Count lines of a file in the deleted worktree via git show."""
    try:
        r = subprocess.run(
            ["git", "-C", GIT_REPO, "show", f"{WORKTREE_COMMIT}:{relpath}"],
            capture_output=True, text=True, timeout=15, check=False,
        )
        if r.returncode == 0:
            return max(1, r.stdout.count("\n"))
        return None
    except (OSError, subprocess.TimeoutExpired):
        return None


def _count_lines_disk(abspath: str) -> int | None:
    """Count lines of a file that exists on disk."""
    try:
        with open(abspath, "rb") as fh:
            return max(1, fh.read().count(b"\n"))
    except OSError:
        return None


def count_lines(path: str) -> int | None:
    """Count lines of a file. Cached."""
    if path not in _LEN_CACHE:
        if path.startswith(WORKTREE_PREFIX):
            rel = path[len(WORKTREE_PREFIX):]
            _LEN_CACHE[path] = _count_lines_worktree(rel)
        elif Path(path).exists():
            _LEN_CACHE[path] = _count_lines_disk(path)
        else:
            _LEN_CACHE[path] = None
    return _LEN_CACHE[path]


# ── helpers ─────────────────────────────────────────────────────────────────

def sha1(s: str) -> str:
    return hashlib.sha1(s.encode("utf-8")).hexdigest()


_sha1_prefix: str | None = None


def _hash_short(value: str) -> str:
    """8-hex-char prefix of sha1."""
    s = sha1(value)
    return s[:8]


def scrub_args(inp: dict[str, Any]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for k, v in inp.items():
        if k in SAFE_KEYS:
            result[k] = v
        elif isinstance(v, str):
            result[k] = _hash_short(v) if len(v) > 0 else v
        elif isinstance(v, (list, dict)):
            result[k] = _hash_short(json.dumps(v, sort_keys=True))
        else:
            result[k] = v
    return result


def extract_root_id(db: sqlite3.Connection, session_id: str) -> str:
    current = session_id
    visited = set()
    while current and current not in visited:
        visited.add(current)
        row = db.execute("SELECT parent_id FROM session WHERE id = ?", (current,)).fetchone()
        if row and row["parent_id"]:
            current = row["parent_id"]
        else:
            break
    return current


def extract_title_and_agent(db: sqlite3.Connection, session_id: str) -> tuple:
    """Return (title, agent) for a session. Agent from the first user message."""
    row = db.execute("SELECT title FROM session WHERE id = ?", (session_id,)).fetchone()
    title = row["title"] if row and row["title"] else ""

    agent = ""
    msg_rows = db.execute(
        """SELECT data FROM message
           WHERE session_id = ?
           ORDER BY time_created""",
        (session_id,),
    ).fetchall()

    for mr in msg_rows:
        mdata = json.loads(mr["data"])
        if mdata.get("role") == "user":
            agent = mdata.get("agent", "")
            break

    return title, agent


# ── file_lines (MAP {path: line_count or null}) ──────────────────────────────

def extract_file_lines(db: sqlite3.Connection, session_id: str) -> dict[str, int | None]:
    """Collect all file paths targeted by read/bash calls and resolve line counts."""
    rows = db.execute(
        """SELECT p.data FROM part p
           WHERE p.session_id = ? AND json_extract(p.data, '$.type') = 'tool'
           ORDER BY p.time_created""",
        (session_id,),
    ).fetchall()

    paths: dict[str, int] = {}  # path -> order seen (to preserve insertion order)
    order = 0

    for r in rows:
        data = json.loads(r["data"])
        tool = data.get("tool", "")
        state = data.get("state", {})
        if not isinstance(state, dict):
            continue
        inp = state.get("input", {})
        if not isinstance(inp, dict):
            continue

        fpath = None
        if tool in ("read", "edit", "write"):
            fpath = inp.get("filePath")
        elif tool in ("bash", "interactive_bash"):
            cmd = inp.get("command") or inp.get("tmux_command") or ""
            for m in _PATH_RE.findall(cmd):
                fpath = m
                break

        if fpath and isinstance(fpath, str) and fpath not in paths:
            paths[fpath] = order
            order += 1

    result: dict[str, int | None] = {}
    for p in sorted(paths, key=paths.get):
        result[p] = count_lines(p)

    return result


# ── calls (landed is BOOLEAN per prototype) ─────────────────────────────────

def extract_calls(db: sqlite3.Connection, session_id: str) -> list[dict[str, Any]]:
    rows = db.execute(
        """SELECT p.data FROM part p
           WHERE p.session_id = ? AND json_extract(p.data, '$.type') = 'tool'
           ORDER BY p.time_created""",
        (session_id,),
    ).fetchall()

    calls: list[dict[str, Any]] = []

    for r in rows:
        data = json.loads(r["data"])
        tool = data.get("tool", "")
        state = data.get("state", {})
        if not isinstance(state, dict):
            continue

        inp = state.get("input", {})
        if not isinstance(inp, dict):
            inp = {}

        md = state.get("metadata", {})
        if not isinstance(md, dict):
            md = {}

        ts_raw = state.get("time", {})
        ts = ts_raw.get("start", 0) if isinstance(ts_raw, dict) else 0

        # prototype:44-45
        landed = (
            tool in ("edit", "write", "patch") and bool(md.get("diff"))
        ) or (
            tool == "bash"
            and "git commit" in (inp.get("command") or "")
            and md.get("exit") == 0
        )

        calls.append({
            "t": ts,
            "tool": tool,
            "args": scrub_args(inp),
            "landed": landed,
        })

    return calls


# ── assembly ────────────────────────────────────────────────────────────────

def build_fixture() -> list[dict[str, Any]]:
    if not DB_PATH.exists():
        print(f"Error: DB not found at {DB_PATH}", file=sys.stderr)
        sys.exit(1)

    db = sqlite3.connect(str(DB_PATH))
    db.row_factory = sqlite3.Row

    fixture: list[dict[str, Any]] = []
    errors: list[str] = []

    for session_id, label in LABELS.items():
        row = db.execute("SELECT id, parent_id FROM session WHERE id = ?", (session_id,)).fetchone()
        if not row:
            errors.append(f"Session not found in DB: {session_id}")
            continue

        title, agent = extract_title_and_agent(db, session_id)
        root_id = extract_root_id(db, session_id)
        file_lines = extract_file_lines(db, session_id)
        calls = extract_calls(db, session_id)

        fixture.append({
            "session_id": session_id,
            "label": label,
            "root_id": root_id,
            "title": title,
            "agent": agent,
            "file_lines": file_lines,
            "calls": calls,
        })

    db.close()

    if errors:
        for err in errors:
            print(err, file=sys.stderr)
        if not fixture:
            sys.exit(1)

    return fixture


def main():
    fixture = build_fixture()
    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    with open(OUTPUT_PATH, "w") as f:
        json.dump(fixture, f, indent=None, separators=(",", ":"), ensure_ascii=False)

    total_calls = sum(len(e["calls"]) for e in fixture)
    total_fl = sum(len(e["file_lines"]) for e in fixture)
    size_kb = OUTPUT_PATH.stat().st_size / 1024
    mf = sum(1 for e in fixture if e["label"] == "must_flag")
    mn = sum(1 for e in fixture if e["label"] == "must_not_flag")

    print(f"Exported {len(fixture)} sessions ({mf} must_flag, {mn} must_not_flag)")
    print(f"  Total tool calls: {total_calls}")
    print(f"  Total file_lines entries: {total_fl}")
    print(f"  File size: {size_kb:.1f} KB")
    print(f"  Output: {OUTPUT_PATH}")

    print(f"\n{'session':36s} {'label':15s} {'title':60s} {'agent':15s} {'calls':>6s} {'landed':>7s} fl_entries")
    print("-" * 150)
    for e in fixture:
        sid = e["session_id"][:35]
        landed_true = sum(1 for c in e["calls"] if c["landed"])
        nonnull_fl = sum(1 for v in e["file_lines"].values() if v is not None)
        print(f"{sid:36s} {e['label']:15s} {e['title'][:60]:60s} {e['agent']:15s} {len(e['calls']):6d} {landed_true:7d} {len(e['file_lines']):3d} ({nonnull_fl} with count)")


if __name__ == "__main__":
    main()