WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-22 01:24:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] browser: instrument skill lifecycle, measure replay speedup, flag silent non-help

Browse Source
This commit is contained in:
ciregenz
2026-06-02 18:03:43 -07:00
parent 294e96d782
commit 2b2d5909a8
2 changed files with 198 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_skill_metrics_and_analyzer.py
+135
View File
@@ -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"]
scripts/analyze-browser-metrics.py
+63 -1
View File
@@ -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()