6krrt/plans/no-progress-detection-prototype.py

#!/usr/bin/env python3
"""Interim opencode loop watcher (stand-in until the no-progress lift ships).

Signals per session, over a sliding window of its last WINDOW tool calls:
  dup    share of calls that repeat an earlier (tool, args) in the window
  top    count of the most repeated (tool, args) in the window
  landed any edit/write with a non-empty diff, or a `git commit` with exit 0,
         by the session OR any descendant within the window's time span

Flag when the window is full enough and nothing landed and
  dup >= DUP_MIN  or  top >= TOP_MIN.
Read-only agents (explore/librarian/oracle) never land, so for them only
top >= TOP_MIN_RO counts.

Modes:
  backtest  : replay every session's history, report worst window per session
  watch     : poll live sessions; on a new flag print it, notify-send, exit 0
"""
import json, sys, time, subprocess, urllib.request, collections, os

U = os.environ.get("OC_URL", "http://127.0.0.1:4097")
WINDOW, MIN_CALLS = 60, 40
DUP_MIN, TOP_MIN, TOP_MIN_RO = 0.25, 12, 8
CUM_MIN = 15  # one exact call repeated this often over the whole session: a slow loop
COVER_MIN = 4.0  # lines read from one file in the window / its length: re-reading, whatever the offsets
READ_ONLY = ("@explore", "@librarian", "@oracle", "explore subagent")


def get(path):
    return json.load(urllib.request.urlopen(U + path, timeout=30))


def calls_of(sid):
    out = []
    for m in get(f"/session/{sid}/message"):
        for p in m["parts"]:
            if p.get("type") != "tool":
                continue
            st = p.get("state") or {}
            inp = st.get("input") or {}
            md = st.get("metadata") or {}
            t = (st.get("time") or {}).get("start") or m["info"].get("time", {}).get("created", 0)
            tool = p.get("tool")
            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)
            out.append((t, tool, json.dumps(inp, sort_keys=True), landed))
    return out


import re
_PATHISH = re.compile(r"[\w./-]+\.(?:json|md|py|js|html|yaml|yml|toml|txt|sql|sh|mjs|css)\b")


_CD_PREFIX = re.compile(r"^\s*cd\s+\S+\s*(?:&&|;)\s*")
_ENV_PREFIX = re.compile(r"^\s*(?:[A-Z_][A-Z0-9_]*=\S+\s+)+")


def _normalize_bash(cmd):
    """Drop comment lines and leading `cd dir &&` / `VAR=x` prefixes, so
    differently-worded commands under one prefix do not collapse to one key."""
    lines = [ln for ln in cmd.splitlines() if not ln.lstrip().startswith("#")]
    c = " ".join(lines).strip()
    for _ in range(3):
        c2 = _ENV_PREFIX.sub("", _CD_PREFIX.sub("", c))
        if c2 == c:
            break
        c = c2
    return c


def target_of(tool, args_json):
    """Coarse target: the file a call is about, ignoring offsets and the exact
    command wording, so 28 differently-worded `cat boulder.json` count as one."""
    try:
        a = json.loads(args_json)
    except Exception:
        return (tool, args_json[:80])
    if a.get("filePath"):
        # A read keys on its slice: walking a big file in pieces is progress,
        # re-reading the same slice is not.
        return ("file", os.path.basename(a["filePath"]), a.get("offset"), a.get("limit")) if tool == "read" \
            else ("file", os.path.basename(a["filePath"]))
    if tool == "bash":
        cmd = _normalize_bash(a.get("command") or "")
        files = sorted({os.path.basename(f) for f in _PATHISH.findall(cmd)})
        if files:
            # Numbers in the command are the slice (sed -n 100,160p, head -40):
            # walking a file is progress; the same slice again is not.
            nums = tuple(sorted(set(re.findall(r"\b\d+\b", cmd))))
            return ("bash-files", ",".join(files[:3]), nums)
        return ("bash", " ".join(cmd.split()[:2]))
    if tool in ("grep", "glob"):
        return (tool, str(a.get("pattern"))[:60])
    return (tool, args_json[:80])


def window_stats(win):
    fp = collections.Counter((c[1], c[2]) for c in win)
    dup = sum(v - 1 for v in fp.values() if v > 1) / max(1, len(win))
    tg = collections.Counter(target_of(c[1], c[2]) for c in win)
    (top_k, top_n) = tg.most_common(1)[0] if tg else (("", ""), 0)
    return dup, top_n, top_k


_LEN = {}


def file_lines(path):
    if path not in _LEN:
        try:
            with open(path, "rb") as fh:
                _LEN[path] = max(1, fh.read().count(b"\n"))
        except OSError:
            _LEN[path] = None
    return _LEN[path]


def coverage(win):
    """Max over files of (lines requested in the window / file length)."""
    req = collections.Counter()
    for c in win:
        if c[1] != "read":
            continue
        a = json.loads(c[2])
        fp = a.get("filePath")
        n = file_lines(fp) if fp else None
        if not n:
            continue
        req[fp] += min(a.get("limit") or n, n)
    best = max(((v / file_lines(k), k) for k, v in req.items()), default=(0.0, ""))
    return round(best[0], 1), os.path.basename(best[1])


def is_ro(title):
    t = (title or "").lower()
    return any(k in t for k in READ_ONLY)


def evaluate(sess, calls, landed_times_tree, end_idx=None):
    calls = calls if end_idx is None else calls[:end_idx]
    win = calls[-WINDOW:]
    if len(win) < MIN_CALLS:
        return None
    t0, t1 = win[0][0], win[-1][0]
    landed = any(t0 <= lt <= t1 for lt in landed_times_tree)
    dup, top_n, top_k = window_stats(win)
    ro = is_ro(sess.get("title"))
    cum_k, cum_n = collections.Counter((c[1], c[2]) for c in calls).most_common(1)[0]
    slow = (not landed) and cum_n >= CUM_MIN
    cov, cov_file = coverage(win)
    reread = (not landed) and cov >= COVER_MIN
    if reread and not (top_n >= TOP_MIN):
        top_n, top_k = int(cov), ("coverage", f"{cov_file} read {cov}x its length")
    if ro:
        flag = top_n >= TOP_MIN_RO or slow or reread
    else:
        flag = ((not landed) and (dup >= DUP_MIN or top_n >= TOP_MIN)) or slow or reread
    if slow and not (top_n >= TOP_MIN):
        top_n, top_k = cum_n, ("exact-total", cum_k[0], cum_k[1][:80])
    return dict(flag=flag, dup=round(dup, 2), top=top_n, top_what=" ".join(str(x) for x in top_k)[:120],
                landed=landed, ro=ro, n=len(calls), t1=t1)


def tree_landed(sessions, calls_by):
    kids = collections.defaultdict(list)
    for s in sessions:
        if s.get("parentID"):
            kids[s["parentID"]].append(s["id"])
    memo = {}
    def lt(sid):
        if sid in memo:
            return memo[sid]
        own = [c[0] for c in calls_by.get(sid, []) if c[3]]
        for k in kids.get(sid, []):
            own += lt(k)
        memo[sid] = own
        return own
    return lt


def backtest(since_ms):
    sessions = [s for s in get("/session") if s.get("time", {}).get("updated", 0) >= since_ms]
    calls_by = {s["id"]: calls_of(s["id"]) for s in sessions}
    lt = tree_landed(sessions, calls_by)
    for s in sessions:
        cs = calls_by[s["id"]]
        worst, first_flag = None, None
        for i in range(MIN_CALLS, len(cs) + 1, 5):
            r = evaluate(s, cs, lt(s["id"]), i)
            if r and (worst is None or (r["flag"], r["dup"], r["top"]) > (worst["flag"], worst["dup"], worst["top"])):
                worst = r
            if r and r["flag"] and first_flag is None:
                first_flag = i
        if worst:
            print(f"{'FLAG' if worst['flag'] else 'ok  '} calls={len(cs):4d} first_flag_at={first_flag} "
                  f"dup={worst['dup']} top={worst['top']} landed={worst['landed']} ro={worst['ro']} | {s.get('title','')[:50]}"
                  + (f"\n        top: {worst['top_what'][:110]}" if worst['flag'] else ""))


def watch(minutes, state_path):
    seen = set()
    if os.path.exists(state_path):
        seen = set(json.load(open(state_path)))
    deadline = time.time() + minutes * 60
    while time.time() < deadline:
        try:
            now = time.time() * 1000
            sessions = get("/session")
            status = get("/session/status")
            recent = [s for s in sessions if s["id"] in status or now - s.get("time", {}).get("updated", 0) < 5 * 60e3]
            calls_by = {s["id"]: calls_of(s["id"]) for s in recent}
            # Judge only sessions doing work now: busy, or a tool call in the
            # last 5 min. A session's "updated" stamp moves on abort, restart
            # and title edits, so it cannot stand in for activity.
            active = [s for s in recent if s["id"] in status
                      or (calls_by[s["id"]] and now - calls_by[s["id"]][-1][0] < 5 * 60e3)]
            lt = tree_landed(sessions, calls_by)
            for s in active:
                r = evaluate(s, calls_by[s["id"]], lt(s["id"]))
                key = f"{s['id']}:{len(calls_by[s['id']]) // 60}"
                if r and r["flag"] and key not in seen:
                    seen.add(key)
                    json.dump(sorted(seen), open(state_path, "w"))
                    msg = (f"{s.get('title','')[:60]} ({s['id']}): {r['n']} calls, dup {r['dup']}, "
                           f"top x{r['top']}: {r['top_what']}, landed={r['landed']}, busy={s['id'] in status}")
                    print("LOOP SUSPECT:", msg)
                    subprocess.run(["notify-send", "-u", "critical", "opencode loop suspected", msg[:250]],
                                   check=False, timeout=5)
                    return 0
        except Exception as e:  # opencode down or restarting: keep watching
            print(f"[{time.strftime('%H:%M:%S')}] poll error: {e}", file=sys.stderr)
        time.sleep(120)
    print("no loops seen in", minutes, "min")
    return 0


if __name__ == "__main__":
    if sys.argv[1] == "backtest":
        backtest(int(sys.argv[2]))
    else:
        sys.exit(watch(int(sys.argv[2]), sys.argv[3]))