#!/usr/bin/env python3
"""Agent Runaway Check (free, SkillHammer) v1.0.0

A 1-minute, read-only check for the settings and habits behind runaway Claude Code / Cursor bills:

  SETTINGS  Claude Code bypassPermissions mode, blanket Bash permission, permission to launch other agents,
            --dangerously-skip-permissions aliases, Cursor CLI approvalMode "unrestricted", Cursor CLI permission
            to start agents from the shell, and whether any hook could stop a runaway session.
  HISTORY   Your heaviest Claude Code sessions in the last N days from local transcripts (~/.claude/projects),
            counted once per API response, with FAN-OUT (many subagents) and LOOP (same tool call repeated) flags.

  python3 runaway_check.py                       checks ~ and the current folder, last 14 days
  python3 runaway_check.py --project ~/code/app --days 30
  python3 runaway_check.py --json

Python 3.9+, standard library only. Reads files; changes nothing; nothing leaves your machine.
"Weighted tokens" = input + 1.25 x cache writes + 0.1 x cache reads + 5 x output: a relative cost measure, not your plan's meter.
"""
import argparse
import collections
import datetime as dt
import glob
import hashlib
import json
import os
import re
import sys

VERSION = "1.0.0"
W = {"input_tokens": 1.0, "cache_creation_input_tokens": 1.25, "cache_read_input_tokens": 0.1, "output_tokens": 5.0}
AGENT_CLIS = r"(claude|cursor-agent|agent|codex|gemini|opencode|aider)"
HIGH, MED, INFO = "HIGH", "MEDIUM", "INFO"
PAID = "Agent Spend Guard (SkillHammer) enforces subagent caps, blocks nested agents and retry loops, and stops sessions at a token budget."


def load_json(path):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f), None
    except FileNotFoundError:
        return None, None
    except (OSError, ValueError) as e:
        return None, str(e)


def finding(sev, rid, where, msg, fix):
    return {"severity": sev, "id": rid, "where": where, "message": msg, "fix": fix}


def check_claude(home, project):
    out, hooks_seen = [], False
    files = [os.path.join(home, ".claude", "settings.json")]
    if project:
        files += [os.path.join(project, ".claude", "settings.json"), os.path.join(project, ".claude", "settings.local.json")]
    for p in files:
        data, err = load_json(p)
        if err:
            out.append(finding(MED, "CC-SETTINGS-INVALID", p, f"can't parse: {err}", "Fix the JSON; Claude Code may be ignoring this file."))
        if not isinstance(data, dict):
            continue
        perms = data.get("permissions") or {}
        if perms.get("defaultMode") == "bypassPermissions":
            out.append(finding(HIGH, "CC-BYPASS", p, "permissions.defaultMode is bypassPermissions: every tool call runs without asking",
                               "Use \"acceptEdits\" or the default mode, and allow specific commands instead."))
        for rule in perms.get("allow") or []:
            r = str(rule).strip()
            if r in ("Bash", "Bash(*)", "Bash(:*)"):
                out.append(finding(MED, "CC-BASH-ANY", p, f"allow rule {r!r} lets the agent run any shell command unprompted",
                                   "Allow specific commands, e.g. Bash(npm test:*), Bash(git status)."))
            elif re.match(r"^Bash\(\s*(npx\s+(-y\s+)?)?" + AGENT_CLIS + r"\b", r):
                out.append(finding(HIGH, "CC-NESTED-ALLOWED", p, f"allow rule {r!r} lets the agent start another agent session unprompted",
                                   "Remove it. Agents launching agents from the shell is how sessions multiply."))
        if data.get("hooks"):
            hooks_seen = True
    return out, hooks_seen


def check_cursor(home, project):
    out, hooks_seen = [], False
    # Cursor CLI reads ~/.cursor/cli-config.json (global) and <project>/.cursor/cli.json (project permissions).
    # Checked against cursor-agent 2026.10.22-27dcf5f and cursor.com/docs/cli/reference/permissions on 30 Oct 2026.
    configs = [os.path.join(home, ".cursor", "cli-config.json")] + ([os.path.join(project, ".cursor", "cli.json")] if project else [])
    for p in configs:
        data, err = load_json(p)
        if err:
            out.append(finding(MED, "CURSOR-CONFIG-INVALID", p, f"can't parse: {err}", "Fix the JSON."))
        if not isinstance(data, dict):
            continue
        if str(data.get("approvalMode", "")).lower() == "unrestricted":
            out.append(finding(HIGH, "CURSOR-UNRESTRICTED", p, "Cursor CLI approvalMode is \"unrestricted\" (Run Everything): shell commands run without approval",
                               "Set \"approvalMode\": \"allowlist\" (what Cursor staff suggested after a recursive-spawn incident, forum 17 Oct 2026)."))
        for rule in ((data.get("permissions") or {}).get("allow") or []):
            r = str(rule).strip()
            if re.match(r"^Shell\(\s*\*(:\*)?\s*\)$", r):
                out.append(finding(MED, "CURSOR-SHELL-ANY", p, f"allow rule {r!r} lets the agent run any shell command unprompted",
                                   "Allow specific commands, e.g. Shell(git), Shell(npm)."))
            elif re.match(r"^Shell\(\s*(npx\s+(-y\s+)?)?" + AGENT_CLIS + r"\b", r):
                out.append(finding(HIGH, "CURSOR-NESTED-ALLOWED", p, f"allow rule {r!r} lets the agent start another agent from the shell",
                                   "Remove it; start agents yourself."))
    for hp in [os.path.join(home, ".cursor", "hooks.json")] + ([os.path.join(project, ".cursor", "hooks.json")] if project else []):
        h, _ = load_json(hp)
        if isinstance(h, dict) and h.get("hooks"):
            hooks_seen = True
    return out, hooks_seen


def check_shell_rc(home):
    out = []
    for name in (".bashrc", ".zshrc", ".bash_profile", ".profile", ".config/fish/config.fish"):
        p = os.path.join(home, name)
        try:
            with open(p, encoding="utf-8", errors="replace") as f:
                for n, line in enumerate(f, 1):
                    if line.lstrip().startswith("#"):
                        continue
                    if "--dangerously-skip-permissions" in line or re.search(r"--permission-mode[ =]['\"]?bypassPermissions", line):
                        out.append(finding(HIGH, "SHELL-SKIP-PERMISSIONS", f"{p}:{n}", "an alias or script always starts Claude Code with --dangerously-skip-permissions",
                                           "Keep it for throwaway containers only; don't make it your default."))
                    # `agent` is the same Cursor CLI as `cursor-agent`; --yolo is an alias of --force (cursor-agent --help, 2026.10.22)
                    if re.search(r"(^|[\s'\"=;|&(])(cursor-agent|agent)\s[^#\n]*(--force|--yolo|-f\b)", line):
                        out.append(finding(MED, "SHELL-CURSOR-FORCE", f"{p}:{n}", "an alias always runs the Cursor CLI with --force/--yolo (commands run without approval)",
                                           "Drop --force/--yolo from the default alias."))
        except OSError:
            continue
    return out


def ts(s):
    try:
        return dt.datetime.fromisoformat(str(s).replace("Z", "+00:00"))
    except ValueError:
        return None


def history(home, days, now=None):
    root = os.path.join(home, ".claude", "projects")
    now = now or dt.datetime.now(dt.timezone.utc)
    since = now - dt.timedelta(days=days)
    seen = set()
    sessions = collections.defaultdict(lambda: {"weighted": 0.0, "responses": 0, "subagents": set(), "tools": collections.Counter(), "project": "", "first": None, "last": None})
    files = glob.glob(os.path.join(root, "*", "*.jsonl")) + glob.glob(os.path.join(root, "*", "*", "subagents", "*.jsonl"))
    for path in files:
        rel = os.path.relpath(path, root).split(os.sep)
        project, sub = rel[0], len(rel) > 2
        sid = rel[1] if sub else rel[1][:-6]
        try:
            f = open(path, encoding="utf-8", errors="replace")
        except OSError:
            continue
        with f:
            for line in f:
                try:
                    o = json.loads(line)
                except ValueError:
                    continue
                if o.get("type") != "assistant":
                    continue
                t = ts(o.get("timestamp"))
                if not t or t < since:
                    continue
                msg = o.get("message") or {}
                s = sessions[o.get("sessionId") or sid]
                s["project"] = project
                if sub:
                    s["subagents"].add(rel[-1])
                mid = msg.get("id")
                if not mid or mid in seen:
                    continue
                seen.add(mid)
                for b in msg.get("content") or []:
                    if isinstance(b, dict) and b.get("type") == "tool_use":
                        key = b.get("name", "") + hashlib.sha1(json.dumps(b.get("input"), sort_keys=True).encode()).hexdigest()[:12]
                        s["tools"][key] += 1
                u = msg.get("usage") or {}
                s["weighted"] += sum(float(u.get(k) or 0) * v for k, v in W.items())
                s["responses"] += 1
                s["first"] = min(filter(None, [s["first"], t]))
                s["last"] = max(filter(None, [s["last"], t]))
    rows = []
    for sid, s in sessions.items():
        loops = [k for k, n in s["tools"].items() if n >= 5]
        flags = (["FAN-OUT"] if len(s["subagents"]) >= 8 else []) + (["LOOP"] if loops else [])
        rows.append({"session": sid, "project": s["project"], "weighted_tokens": round(s["weighted"]), "responses": s["responses"],
                     "subagents": len(s["subagents"]), "max_identical_tool_calls": max(s["tools"].values(), default=0), "flags": flags,
                     "start": s["first"].isoformat() if s["first"] else None})
    rows.sort(key=lambda r: -r["weighted_tokens"])
    return {"root": root, "found": os.path.isdir(root), "days": days, "sessions": rows, "total_weighted": sum(r["weighted_tokens"] for r in rows)}


def run(home, project, days, now=None):
    f1, cc_hooks = check_claude(home, project)
    f2, cu_hooks = check_cursor(home, project)
    findings = f1 + f2 + check_shell_rc(home)
    if not (cc_hooks or cu_hooks):
        findings.append(finding(INFO, "NO-HOOKS", project or home, "no Claude Code or Cursor hooks configured: nothing can stop a runaway session automatically",
                                "Add hooks that cap subagents, block nested agents and stop at a budget."))
    hist = history(home, days, now)
    if any(r["flags"] for r in hist["sessions"]):
        findings.append(finding(MED, "HISTORY-FLAGS", hist["root"], "sessions with FAN-OUT or LOOP patterns in your history (see below)",
                                "Cap concurrent subagents and stop identical retries."))
    order = {HIGH: 0, MED: 1, INFO: 2}
    findings.sort(key=lambda x: (order[x["severity"]], x["id"]))
    return {"version": VERSION, "findings": findings, "history": hist}


def human(n):
    return f"{n/1e6:.1f}M" if n >= 1e6 else f"{n/1e3:.0f}k" if n >= 1e3 else str(int(n))


def text(r, top=5):
    out = [f"Agent Runaway Check v{VERSION}", ""]
    hi = sum(1 for f in r["findings"] if f["severity"] == HIGH)
    out.append(f"SETTINGS: {len(r['findings'])} finding(s), {hi} high")
    for f in r["findings"]:
        out += [f"  [{f['severity']}] {f['id']}  {f['where']}", f"      {f['message']}", f"      fix: {f['fix']}"]
    if not r["findings"]:
        out.append("  nothing risky found")
    h = r["history"]
    out += ["", f"HISTORY (last {h['days']} days, Claude Code transcripts)"]
    if not h["found"]:
        out.append(f"  no transcripts at {h['root']}")
    elif not h["sessions"]:
        out.append("  no sessions in this period")
    else:
        out.append(f"  {len(h['sessions'])} session(s), {human(h['total_weighted'])} weighted tokens. Heaviest:")
        for s in h["sessions"][:top]:
            share = 100 * s["weighted_tokens"] / h["total_weighted"] if h["total_weighted"] else 0
            out.append(f"   {human(s['weighted_tokens']):>7} ({share:4.1f}%)  {s['session'][:8]}  {s['project'][:40]}  subagents {s['subagents']}  max repeats {s['max_identical_tool_calls']}  {' '.join(s['flags'])}")
    out += ["", "Free fixes are listed above. " + PAID, "https://skillhammer.com"]
    return "\n".join(out)


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--project", default=os.getcwd())
    ap.add_argument("--home", default=os.path.expanduser("~"), help=argparse.SUPPRESS)
    ap.add_argument("--days", type=int, default=14)
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--version", action="version", version=VERSION)
    a = ap.parse_args(argv)
    r = run(a.home, os.path.abspath(a.project), a.days)
    print(json.dumps(r, indent=2) if a.json else text(r))
    return 1 if any(f["severity"] == HIGH for f in r["findings"]) else 0


if __name__ == "__main__":
    sys.exit(main())
