6krrt/scripts/progress_backtest.py

#!/usr/bin/env python
"""Progress-detection backtest: replay sessions through the detector.

Reads sessions from a fixture JSON file (default ~/.local/share/opencode/rc-
servers.json for live mode, or --fixture <path>) and runs
progress_detect.evaluate() over a sliding window with step 5.

Output per session:
    FLAG|ok  calls=NNNN first_flag_at=N dup=0.00 top=N landed=0 ro=N | session_title

Usage:
    # Live mode — reads sessions from the opencode rc-servers.json
    PYTHONPATH=src python -m scripts.progress_backtest

    # Fixture mode
    PYTHONPATH=src python -m scripts.progress_backtest --fixture /path/to/fixture.json
"""

from __future__ import annotations

import argparse
import json
import os
import sys

from progress_detect import DetectConfig, evaluate

_DEFAULT_CFG = DetectConfig()
_FIXTURE_PATH = os.path.join(
    os.path.dirname(__file__), "..", "tests", "fixtures", "progress", "fixture.json"
)
_RC_SERVERS = os.path.expanduser("~/.local/share/opencode/rc-servers.json")


def _calls_from_fixture(session):
    """Convert a fixture session dict to call tuples for evaluate()."""
    return [
        (c["t"], c["tool"], json.dumps(c["args"]), c["landed"])
        for c in session["calls"]
    ]


def _file_lines_fn(file_lines_dict):
    """Return a callable for coverage() from a file_path→line_count dict."""
    def fn(fp):
        if fp is None:
            return 0
        return file_lines_dict.get(fp, 0) or 0
    return fn


def _backtest_session(session, cfg):
    """Run sliding-window backtest on one session. Returns (first_flag_at, reason_dict)."""
    calls = _calls_from_fixture(session)
    fl = session.get("file_lines", {})
    fl_fn = _file_lines_fn(fl)
    title = session.get("title", "")

    if len(calls) < cfg.min_calls:
        return (None, None)

    for end_idx in range(cfg.min_calls, len(calls) + 1, 5):
        flag, reason = evaluate(calls, cfg, fl_fn, title, end_idx=end_idx)
        if flag and reason:
            return (end_idx, reason)
    return (None, None)


def _format_line(session, first_flag_at, reason):
    """Format one backtest output line.

    Pattern: FLAG|ok  calls=NNNN first_flag_at=N dup=0.00 top=N landed=ro=N | title
    """
    calls_n = len(session["calls"])
    title = session.get("title", "")

    if reason is not None:
        dup = reason.get("dup", 0.0)
        top = reason.get("top", 0)
        landed = 1 if reason.get("landed") else 0
        ro = 1 if reason.get("ro") else 0
        first_flag = first_flag_at
        flags = "FLAG  "
    else:
        dup = 0.0
        top = 0
        landed = 0
        ro = 0
        first_flag = "N/A"
        flags = "ok    "

    return (
        f"{flags}calls={calls_n:4d} first_flag_at={first_flag}"
        f" dup={dup:.2f} top={top}"
        f" landed={landed} ro={ro}"
        f" | {title}"
    )


def _load_fixture(path):
    """Load a fixture JSON file and return list of session dicts."""
    with open(path) as f:
        return json.load(f)


def _load_live_sessions():
    """Load sessions from the rc-servers.json opencode API file."""
    if not os.path.isfile(_RC_SERVERS):
        print(f"Error: {_RC_SERVERS} not found. Use --fixture instead.", file=sys.stderr)
        sys.exit(1)

    with open(_RC_SERVERS) as f:
        data = json.load(f)

    sessions = []
    # rc-servers.json structure: {<session_id>: {<agent>: {calls: [...]}}}
    for session_id, agents in data.items():
        for agent_name, agent_data in agents.items():
            calls = agent_data.get("calls", [])
            sessions.append({
                "session_id": session_id,
                "title": agent_data.get("title", f"{agent_name} session"),
                "agent": agent_name,
                "calls": calls,
                "file_lines": {},
                "label": "unknown",
            })
    return sessions


def main():
    parser = argparse.ArgumentParser(
        description="Progress-detection backtest: replay sessions through the detector."
    )
    parser.add_argument(
        "--fixture",
        default=None,
        help="Path to fixture JSON file (default: live mode from rc-servers.json)",
    )
    parser.add_argument(
        "--window",
        type=int,
        default=60,
        help="Sliding window size in calls (default: 60)",
    )
    parser.add_argument(
        "--min-calls",
        type=int,
        default=40,
        help="Minimum calls before evaluation (default: 40)",
    )
    parser.add_argument(
        "--step",
        type=int,
        default=5,
        help="Sliding window step (default: 5)",
    )
    parser.add_argument(
        "--output",
        type=str,
        default=None,
        help="Write output to file instead of stdout",
    )
    args = parser.parse_args()

    cfg = DetectConfig(
        window=args.window,
        min_calls=args.min_calls,
        read_only_agent_keywords=("explore", "librarian", "oracle"),
    )

    if args.fixture:
        fixture_path = args.fixture
    else:
        fixture_path = _FIXTURE_PATH

    if not os.path.isfile(fixture_path):
        print(f"Error: fixture not found at {fixture_path}", file=sys.stderr)
        sys.exit(1)

    sessions = _load_fixture(fixture_path)

    results = []
    for session in sessions:
        first_flag_at, reason = _backtest_session(session, cfg)
        line = _format_line(session, first_flag_at, reason)
        results.append(line)

    output = "\n".join(results) + "\n"

    if args.output:
        os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
        with open(args.output, "w") as f:
            f.write(output)
    else:
        sys.stdout.write(output)

    # Print summary
    flagged = sum(1 for r in results if r.startswith("FLAG"))
    total = len(results)
    print(f"\n# backtest complete: {flagged}/{total} flagged", file=sys.stderr)


if __name__ == "__main__":
    main()