diff --git a/backend/tests/test_skill_metrics_and_analyzer.py b/backend/tests/test_skill_metrics_and_analyzer.py new file mode 100644 index 00000000..cd3d51c8 --- /dev/null +++ b/backend/tests/test_skill_metrics_and_analyzer.py @@ -0,0 +1,135 @@ +"""The skill layer's own honesty check. + +Drives the REAL skill + metrics functions through full multi-run lifecycles and +then runs the REAL analyzer over the emitted JSONL, asserting it (a) measures the +replay speedup when the layer helps and (b) FLAGS the silent ghost when a task is +repeated but never reaches the fast path (thrash / won't-distill). If the analyzer +couldn't tell those apart, "it completed" would hide a feature that never helps. +""" + +import importlib.util +import os +import time + +from backend.apps.agents.browser import browser_skills as sk +from backend.apps.agents.browser import browser_metrics as bm + +_ANALYZER = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "analyze-browser-metrics.py") + + +def _load_analyzer(): + spec = importlib.util.spec_from_file_location("bma", _ANALYZER) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def _log(): + return [ + {"tool": "BrowserNavigate", "input": {"url": "http://h/form"}, "ok": True}, + {"tool": "BrowserType", "input": {"selector": "#q", "text": "shoes"}, "ok": True}, + {"tool": "BrowserClickIndex", "input": {}, "ok": True, "clicked_role": "button", "clicked_name": "Search"}, + ] + + +def _task_row(sig, path, dur_s): + # started_at in the past makes record_task compute a realistic total_ms. + bm.record_task("s-" + sig + path, "b", sig, "completed", + time.time() - dur_s, 0 if path == "replay" else 3, + _log(), {"input": 10, "output": 5}, path=path, task_sig=sig) + + +def test_skill_events_are_emitted_for_each_transition(_metrics_dir): + sk.clear(wipe_disk=True) + sk.record_skill("shop.com", "search now", _log()) # learn + sk.mark_replay_succeeded("shop.com", "search now") # promote + sk.mark_replay_failed("shop.com", "search now") # kept (trusted, 1) + sk.mark_replay_failed("shop.com", "search now") # demote + evs = _read(os.path.join(_metrics_dir, "skill_events.jsonl")) + kinds = [e["kind"] for e in evs] + assert "learn" in kinds and "promote" in kinds and "demote" in kinds + # every event carries enough to group + reason about it + assert all(e.get("host") and e.get("task_sig") and e.get("kind") for e in evs) + + +def test_analyzer_measures_replay_speedup_when_the_layer_helps(_metrics_dir, capsys): + sk.clear(wipe_disk=True) + # A repeated task: 1 slow LLM run, then 2 fast replays -> measurable speedup. + sk.record_skill("shop.com", "search now", _log()) + _task_row(sk._sig("search now"), "llm", 4.0) + sk.mark_replay_succeeded("shop.com", "search now") + _task_row(sk._sig("search now"), "replay", 0.04) + _task_row(sk._sig("search now"), "replay", 0.05) + + mod = _load_analyzer() + tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl")) + sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl")) + mod.skill_layer_report(tasks, sevs) + out = capsys.readouterr().out + assert "REPLAY SPEEDUP" in out + assert "x faster" in out and "replay" in out + + +def test_analyzer_flags_silent_non_help_thrash(_metrics_dir, capsys): + sk.clear(wipe_disk=True) + # A task that keeps getting re-learned/edited and quarantined, never promoted, + # and whose runs always go via the LLM (never the fast path) = the ghost. + sk.record_skill("bad.com", "do thing now", _log()) # learn + sk.mark_replay_failed("bad.com", "do thing now") # quarantine + edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True, + "clicked_role": "button", "clicked_name": "Other"}] + sk.record_skill("bad.com", "do thing now", edited) # edit (un-quarantine) + sk.mark_replay_failed("bad.com", "do thing now") # quarantine again + _task_row(sk._sig("do thing now"), "llm", 3.0) + _task_row(sk._sig("do thing now"), "llm_fallback", 3.2) + + mod = _load_analyzer() + tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl")) + sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl")) + mod.skill_layer_report(tasks, sevs) + out = capsys.readouterr().out + assert "SILENT NON-HELP" in out # repeated but never replayed + assert "THRASH" in out # re-learned/edited, never promoted + + +def test_analyzer_reports_composition(_metrics_dir, capsys): + sk.clear(wipe_disk=True) + sk.record_skill("shop.com", "search now", _log()) + sk.mark_replay_succeeded("shop.com", "search now") # trusted foundation + plus = _log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True, + "clicked_role": "button", "clicked_name": "Checkout"}] + sk.record_skill("shop.com", "search and checkout now", plus) # composes on foundation + sk.mark_replay_succeeded("shop.com", "search and checkout now") # dependent earns trust too + sk.deprecate_skill("shop.com", "search now") # must invalidate the TRUSTED dependent + + mod = _load_analyzer() + sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl")) + # the invalidate EVENT must actually fire (end-to-end), not just the state flip + assert any(e["kind"] == "invalidate" for e in sevs) + mod.skill_layer_report([], sevs) + out = capsys.readouterr().out + assert "composition:" in out + assert "built on a proven sub-skill" in out + assert "1 dependent(s) re-proofed" in out + + +# --- helpers --------------------------------------------------------------- +def _read(path): + import json + out = [] + if os.path.exists(path): + with open(path) as f: + for line in f: + line = line.strip() + if line: + out.append(json.loads(line)) + return out + + +import pytest + + +@pytest.fixture +def _metrics_dir(): + # the autouse conftest fixture already points metrics at a temp dir; surface it + return os.environ["OPENSWARM_BROWSER_METRICS_DIR"] diff --git a/scripts/analyze-browser-metrics.py b/scripts/analyze-browser-metrics.py index 631c5cd3..e374c13e 100755 --- a/scripts/analyze-browser-metrics.py +++ b/scripts/analyze-browser-metrics.py @@ -82,12 +82,72 @@ def ghost_verdict(task, events_for_task): 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 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)}\n") + 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 @@ -146,6 +206,8 @@ def main(): print(f"honest completion rate: {round(100*(completed-ghosts)/n,1)}% " f"(completed minus ghosts)") + skill_layer_report(tasks, skill_events) + if __name__ == "__main__": main()