openswarm/scripts/analyze-browser-metrics.py

#!/usr/bin/env python3
"""
Analyze recorded browser-agent metrics and flag GHOST SUCCESS: tasks that
report "completed" but show no evidence of real work (errors-only, looped, or
zero productive mutation). A task ending without an exception is NOT proof the
goal was achieved, this is the independent reality check.

Usage:
  python3 scripts/analyze-browser-metrics.py [metrics_dir]
Defaults to $OPENSWARM_BROWSER_METRICS_DIR, else
~/Library/Application Support/OpenSwarm/data/browser_metrics (mac) or backend/data/.
"""

import json
import os
import sys
from collections import Counter, defaultdict

# Tools that actually change page state (vs. read/meta). A "completed" task that
# never ran one of these did nothing but look around, suspicious.
PRODUCTIVE = {
    "BrowserClick", "BrowserClickIndex", "BrowserType", "BrowserNavigate",
    "BrowserPressKey", "BrowserBatch", "BrowserReplayRoute",
}


def _default_dir():
    env = os.environ.get("OPENSWARM_BROWSER_METRICS_DIR")
    if env:
        return env
    mac = os.path.expanduser("~/Library/Application Support/OpenSwarm/data/browser_metrics")
    if os.path.isdir(mac):
        return mac
    return os.path.join(os.path.dirname(__file__), "..", "backend", "data", "browser_metrics")


def load(path):
    if not os.path.exists(path):
        return []
    out = []
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    out.append(json.loads(line))
                except Exception:
                    pass
    return out


def ghost_verdict(task, events_for_task):
    """Return (is_ghost, [reasons]). Conservative: only flags 'completed' tasks
    that show no evidence the work happened."""
    if task.get("status") != "completed":
        return False, []
    reasons = []
    tools = [e["tool"] for e in events_for_task]
    productive = [t for t in tools if t in PRODUCTIVE]
    errs = sum(1 for e in events_for_task if not e.get("ok"))
    total = len(events_for_task)
    # A read/extract task legitimately has no state-changing action; its evidence
    # is that a READ tool actually returned content. So "no productive action" is
    # only a ghost when NO read returned data either (i.e. nothing real happened).
    _READ = {"BrowserGetText", "BrowserGetElements", "BrowserListInteractives",
             "BrowserListRoutes", "BrowserReplayRoute", "BrowserScreenshot", "BrowserEvaluate"}
    read_with_content = any(
        e["tool"] in _READ and e.get("ok") and (e.get("result_len", 0) or 0) > 0
        for e in events_for_task
    )
    if total == 0:
        reasons.append("completed with ZERO tool calls (model declared done without acting)")
    if total and not productive and not read_with_content:
        reasons.append("no state-changing action AND no read returned content, yet marked completed")
    if total and errs / total >= 0.5:
        reasons.append(f"{errs}/{total} tool calls errored but still marked completed")
    if any(e.get("is_loop") for e in events_for_task):
        reasons.append("loop detector fired during a 'completed' task")
    prod_ok = [e for e in events_for_task if e["tool"] in PRODUCTIVE and e.get("ok")]
    if productive and not prod_ok:
        reasons.append("every state-changing action errored, yet marked completed")
    return (len(reasons) > 0), reasons


def skill_layer_report(tasks, skill_events):
    """Did the learn/replay/trust layer ACTUALLY help, or is it silently
    thrashing? Measures the replay speedup on repeated tasks and flags the ghost
    where a task is done over and over but never reaches the no-LLM fast path."""
    print("\n=== SKILL LAYER (does learn/replay actually help?) ===")
    paths = Counter(t.get("path", "llm") for t in tasks)
    total = sum(paths.values())
    if total:
        for p in ("replay", "llm", "llm_fallback"):
            if paths.get(p):
                print(f"  {p:<13}{paths[p]:>4}  ({round(100*paths[p]/total)}% of finished tasks)")

    # Repeated tasks: group completed runs by signature, compare replay vs llm time.
    by_sig = defaultdict(list)
    for t in tasks:
        if t.get("completed") and t.get("task_sig"):
            by_sig[t["task_sig"]].append(t)
    repeated = {s: r for s, r in by_sig.items() if len(r) >= 2}
    helped, silent = [], []
    for sig, runs in repeated.items():
        rp = [t["total_ms"] for t in runs if t.get("path") == "replay"]
        lm = [t["total_ms"] for t in runs if t.get("path") in ("llm", "llm_fallback")]
        if rp and lm:
            speed = round((sum(lm) / len(lm)) / max(1, (sum(rp) / len(rp))), 1)
            helped.append((sig, len(runs), speed, round(sum(lm) / len(lm)), round(sum(rp) / len(rp))))
        elif not rp:
            silent.append((sig, len(runs)))
    if helped:
        print("\n  REPLAY SPEEDUP on repeated tasks (the win, measured):")
        for sig, n, speed, lm_ms, rp_ms in sorted(helped, key=lambda x: -x[2]):
            print(f"    {speed}x faster  ({lm_ms}ms LLM -> {rp_ms}ms replay, {n} runs)  {sig[:48]}")
    if silent:
        print("\n  ⚠️ SILENT NON-HELP (task repeated but NEVER hit the fast path):")
        print("     a repeat that never replays = the skill thrashed or won't distill;")
        print("     it still completes, but the speed win never lands. Investigate.")
        for sig, n in sorted(silent, key=lambda x: -x[1]):
            print(f"    x{n}  {sig[:60]}")
    if not helped and not silent:
        print("  (no task repeated yet, so no replay measurement available)")

    if not skill_events:
        return
    # Lifecycle rollup + thrash detector (re-learn loops that never promote).
    kinds = Counter(e.get("kind") for e in skill_events)
    print("\n  lifecycle:", "  ".join(f"{k}={kinds[k]}" for k in
          ("learn", "edit", "promote", "quarantine", "demote", "compose", "invalidate") if kinds.get(k)))
    per = defaultdict(Counter)
    for e in skill_events:
        per[f"{e.get('host')}::{e.get('task_sig')}"][e.get("kind")] += 1
    thrash = [(k, c) for k, c in per.items() if c["learn"] + c["edit"] >= 2 and c["promote"] == 0]
    if thrash:
        print("\n  ⚠️ THRASH (re-learned/edited >=2x but NEVER promoted to trusted):")
        for k, c in thrash:
            print(f"    {k[:60]}  learn={c['learn']} edit={c['edit']} quarantine={c['quarantine']}")
    if kinds.get("compose"):
        print(f"\n  composition: {kinds['compose']} skill(s) built on a proven sub-skill, "
              f"{kinds.get('invalidate', 0)} dependent(s) re-proofed after a foundation changed")


def playbook_report(tasks):
    """Does the tier-2 strategy playbook actually make judgment tasks cheaper over
    time? Compare LLM-path runs on a host BEFORE a playbook existed (cold) vs once
    it was seeded. The win is fewer exploration turns; flag a host where seeded
    runs are NOT cheaper (the 'memory looks active but doesn't help' ghost)."""
    from collections import defaultdict
    by_host = defaultdict(lambda: {"cold": [], "seeded": []})
    for t in tasks:
        if t.get("path") not in ("llm", "llm_fallback") or not t.get("completed"):
            continue
        host = (t.get("task_sig") or "").split(" ")[0] or t.get("browser_id", "?")
        bucket = "seeded" if t.get("playbook_seeded") else "cold"
        by_host[host][bucket].append(t.get("turns", 0) or 0)
    rows = {h: v for h, v in by_host.items() if v["cold"] and v["seeded"]}
    if not any(t.get("playbook_seeded") for t in tasks):
        return  # nothing seeded yet, no measurement to make
    print("\n=== STRATEGIC PLAYBOOK (does learned site-strategy cut exploration?) ===")
    seeded_total = sum(1 for t in tasks if t.get("playbook_seeded"))
    print(f"  runs seeded with a playbook: {seeded_total}")
    if not rows:
        print("  (no host yet has BOTH a cold and a seeded run to compare)")
        return
    for h, v in rows.items():
        cold = sum(v["cold"]) / len(v["cold"])
        seeded = sum(v["seeded"]) / len(v["seeded"])
        verdict = "HELPS" if seeded < cold else "⚠️ NOT HELPING"
        print(f"    {h[:40]:40} cold avg {cold:.1f} turns -> seeded avg {seeded:.1f} turns  {verdict}")


def main():
    d = sys.argv[1] if len(sys.argv) > 1 else _default_dir()
    events = load(os.path.join(d, "events.jsonl"))
    tasks = load(os.path.join(d, "tasks.jsonl"))
    skill_events = load(os.path.join(d, "skill_events.jsonl"))
    print(f"metrics dir: {d}")
    print(f"events: {len(events)}  tasks: {len(tasks)}  skill_events: {len(skill_events)}\n")
    if not tasks and not events:
        print("No metrics recorded yet. Run some browser-agent tasks first.")
        return

    ev_by_session = defaultdict(list)
    for e in events:
        ev_by_session[e.get("session_id")].append(e)

    # Per-tier rollup across all events
    tier_calls, tier_ms, tier_err = Counter(), Counter(), Counter()
    for e in events:
        t = e.get("tier", "other")
        tier_calls[t] += 1
        tier_ms[t] += int(e.get("elapsed_ms", 0) or 0)
        if not e.get("ok"):
            tier_err[t] += 1

    print("=== PER-TIER (across all tool calls) ===")
    print(f"{'tier':<20}{'calls':>6}{'avg_ms':>9}{'err%':>7}")
    for t in sorted(tier_calls, key=lambda x: -tier_calls[x]):
        c = tier_calls[t]
        avg = round(tier_ms[t] / c, 1) if c else 0
        errp = round(100 * tier_err[t] / c, 1) if c else 0
        print(f"{t:<20}{c:>6}{avg:>9}{errp:>6}%")

    print("\n=== PER-TASK (completion, time, cost, ghost check) ===")
    completed = ghosts = 0
    all_errs = Counter()
    for tk in tasks:
        evs = ev_by_session.get(tk.get("session_id"), [])
        is_ghost, reasons = ghost_verdict(tk, evs)
        if tk.get("status") == "completed":
            completed += 1
        if is_ghost:
            ghosts += 1
        for err, n in tk.get("recurring_errors", []):
            all_errs[err] += n
        flag = "  ⚠️ GHOST" if is_ghost else ""
        print(f"- [{tk.get('status')}] {tk.get('total_ms')}ms turns={tk.get('turns')} "
              f"tools={tk.get('tool_calls')} tok_in={tk.get('tokens_in')} "
              f"tok_out={tk.get('tokens_out')} :: {str(tk.get('task'))[:60]}{flag}")
        for r in reasons:
            print(f"      ghost-reason: {r}")

    print("\n=== RECURRING ERRORS (top 10 across tasks) ===")
    for err, n in all_errs.most_common(10):
        print(f"  x{n}  {err}")

    print("\n=== SUMMARY ===")
    n = len(tasks)
    print(f"tasks: {n}  completed: {completed}  ghost-completed: {ghosts}")
    if n:
        avg_ms = round(sum(t.get("total_ms", 0) for t in tasks) / n)
        avg_tok = round(sum(t.get("tokens_in", 0) + t.get("tokens_out", 0) for t in tasks) / n)
        print(f"avg task time: {avg_ms}ms   avg tokens/task: {avg_tok}")
        print(f"honest completion rate: {round(100*(completed-ghosts)/n,1)}% "
              f"(completed minus ghosts)")

    skill_layer_report(tasks, skill_events)
    playbook_report(tasks)


if __name__ == "__main__":
    main()