[eric] browser: version skills behind a trust gate, compose on proven sub-skills

backend/apps/agents/browser/browser_agent.py

@@ -395,7 +395,9 @@ async def run_browser_agent(
     if skill and not concrete_steps:
         logger.info(f"[browser-skills] skill matched on {replay_host} but slots unfillable from task; running full agent")
         skill = None
+    replay_attempted = False
     if skill and concrete_steps:
+        replay_attempted = True
         logger.info(f"[browser-skills] REPLAY attempt: {len(concrete_steps)} steps on {replay_host}")
         replay_log: list[dict] = []
         replay_ok = True
@@ -427,10 +429,11 @@ async def run_browser_agent(
             if res.get("url"):
                 last_seen_url = res["url"]
         if replay_ok and replay_log:
-            browser_skills.mark_replayed(replay_host, task)
+            browser_skills.mark_replay_succeeded(replay_host, task)
             summary = browser_metrics.record_task(
                 session_id, browser_id, task, "completed", metrics_started_at,
                 0, replay_log, session.tokens,
+                path="replay", task_sig=browser_skills._sig(task),
             )
             logger.info(f"[browser-skills] REPLAY SUCCEEDED in {summary['total_ms']}ms with 0 LLM calls")
             try:
@@ -451,7 +454,13 @@ async def run_browser_agent(
                 "action_log": replay_log, "final_screenshot": final_screenshot,
                 "replayed": True,
             }
-        # replay didn't fully succeed -> fall through to the full LLM agent
+        # Replay didn't fully succeed. Update the skill's trust BEFORE falling
+        # through: an unproven skill that failed gets quarantined (never replayed
+        # again -> pure-LLM baseline), a proven one tolerates a transient miss.
+        # The full agent below then re-records edit-aware (new steps -> new rev).
+        if not cancel_event.is_set():
+            verdict = browser_skills.mark_replay_failed(replay_host, task)
+            logger.info(f"[browser-skills] replay fell back to full agent (trust verdict: {verdict})")
 
     text_parts = []  # initialized before loop so post-loop summary (line ~1294) has a default
     # Circuit breaker for ReportProgress violations. Some models get stuck
@@ -687,7 +696,13 @@ async def run_browser_agent(
                     if tu.name == "BrowserListSkills":
                         skills = browser_skills.list_skills(cur_host) if cur_host else []
                         if skills:
-                            lines = "\n".join(f"- \"{s['task']}\" ({s['steps']} steps, reused {s['replays']}x)" for s in skills[:20])
+                            _tag = {"trusted": "proven", "probation": "unproven", "quarantine": "disabled"}
+                            def _fmt_skill(s):
+                                line = f"- \"{s['task']}\" ({s['steps']} steps, {_tag.get(s['state'], s['state'])}, reused {s['replays']}x"
+                                if s.get("builds_on"):
+                                    line += f", builds on {len(s['builds_on'])} other shortcut(s)"
+                                return line + ")"
+                            lines = "\n".join(_fmt_skill(s) for s in skills[:20])
                             meta_text = f"Learned shortcuts for {cur_host}:\n{lines}"
                         else:
                             meta_text = f"No learned shortcuts for {cur_host or 'this site'} yet."
@@ -956,7 +971,9 @@ async def run_browser_agent(
 
         session.status = "completed"
         browser_metrics.record_task(session_id, browser_id, task, "completed",
-                                    metrics_started_at, turn + 1, action_log, session.tokens)
+                                    metrics_started_at, turn + 1, action_log, session.tokens,
+                                    path="llm_fallback" if replay_attempted else "llm",
+                                    task_sig=browser_skills._sig(task))
         # Learn this task: distill the successful run into a replayable skill so
         # the next identical task on this host runs via the no-LLM fast path.
         try:

backend/apps/agents/browser/browser_metrics.py

@@ -8,8 +8,14 @@ recurred. Pure best-effort: every call is wrapped so a metrics failure can
 never break the agent loop.
 
 Files (under DATA_ROOT/browser_metrics/, env-overridable):
-  events.jsonl    one line per tool call
-  tasks.jsonl     one line per finished task (with a recurring-error rollup)
+  events.jsonl        one line per tool call
+  tasks.jsonl         one line per finished task (with a recurring-error rollup)
+  skill_events.jsonl  one line per skill-lifecycle transition (learn / promote /
+                      edit / quarantine / demote / compose / invalidate), so we
+                      can tell whether the skill layer ACTUALLY speeds repeats up
+                      or is silently thrashing (re-learning every run, never
+                      promoting), which is the ghost that "completes" but never
+                      delivers the win.
 """
 
 import json
@@ -101,10 +107,24 @@ def record_tool(session_id, browser_id, turn, tool, elapsed_ms, ok, error,
     )
 
 
+def record_skill_event(kind, host, task_sig, rev=0, state="", extra=None) -> None:
+    """One line per skill-lifecycle transition. Best-effort. `kind` is one of
+    learn / edit / promote / quarantine / demote / compose / invalidate. This is
+    what lets the analyzer prove the skill layer is helping (promotes accumulate,
+    repeats replay) vs. silently thrashing (re-learn loops, never promotes)."""
+    _append("skill_events.jsonl", {
+        "ts": time.time(), "kind": kind, "host": host, "task_sig": task_sig,
+        "rev": rev, "state": state, "extra": extra or {},
+    })
+
+
 def record_task(session_id, browser_id, task, status, started_at, turns,
-                action_log, tokens) -> dict:
+                action_log, tokens, path="llm", task_sig="") -> dict:
     """One summary line per finished task: completion, total time, per-tier
-    latency, token cost, and the recurring-error rollup. Returns the summary."""
+    latency, token cost, and the recurring-error rollup. `path` records HOW the
+    task finished (replay = no-LLM fast path, llm = full agent, llm_fallback =
+    full agent after a replay miss) so we can measure the replay speedup and spot
+    repeats that never reach the fast path. Returns the summary."""
     total_ms = int((time.time() - started_at) * 1000)
     by_tier = {}
     err_counter = Counter()
@@ -125,6 +145,8 @@ def record_task(session_id, browser_id, task, status, started_at, turns,
         "session_id": session_id,
         "browser_id": browser_id,
         "task": (task or "")[:200],
+        "task_sig": task_sig,
+        "path": path,
         "status": status,
         "completed": status == "completed",
         "total_ms": total_ms,
@@ -137,7 +159,7 @@ def record_task(session_id, browser_id, task, status, started_at, turns,
     }
     _append("tasks.jsonl", summary)
     logger.info(
-        f"[browser-metrics] TASK {status} total={total_ms}ms turns={turns} "
+        f"[browser-metrics] TASK {status} path={path} total={total_ms}ms turns={turns} "
         f"tools={len(action_log)} tok_in={summary['tokens_in']} tok_out={summary['tokens_out']} "
         f"recurring_errs={summary['recurring_errors'][:2]}"
     )

backend/apps/agents/browser/browser_skills.py

@@ -24,6 +24,18 @@ Two properties we hold to extreme rigor:
    and never persisted. Only fully non-sensitive skills are written to disk;
    URL userinfo + fragments are stripped before persisting regardless.
 
+3. NOTHING IS TRUSTED UNTIL A REPLAY PROVES IT (the verify gate). A freshly
+   learned or freshly edited skill is PROBATIONARY: it's allowed to replay (that
+   is how it earns trust), but the first time a probationary replay fails it is
+   QUARANTINED, not silently kept; quarantined skills never replay again (the
+   task falls back to the pure-LLM baseline), so a lossy distillation can never
+   make a task slower-than-baseline or ghost-succeed. Only a skill that has
+   replayed end-to-end successfully becomes TRUSTED, and only a trusted skill
+   gets the benefit of the doubt on a one-off transient miss. Re-deriving a task
+   after a failed replay is an EDIT: if the new steps differ from the stored
+   ones the skill is re-versioned (rev++) back to probation; if they're
+   identical the miss was transient and trust is kept. State + rev persist.
+
 Robustness (a stale replay that "succeeds" wrongly is the ghost-failure we must
 avoid): clicks are recorded by (role, name) and re-resolved fresh at replay; a
 skill is only recorded if every productive step is robustly replayable; the
@@ -42,12 +54,34 @@ from urllib.parse import urlparse, urlunparse
 
 logger = logging.getLogger(__name__)
 
+
+def _event(kind: str, host: str, sig: str, rev: int = 0, state: str = "", **extra) -> None:
+    """Mirror a lifecycle transition into the metrics sink so the analyzer can
+    prove the skill layer helps vs. silently thrashes. Lazy + best-effort: this
+    module never hard-depends on metrics, and a metrics failure never propagates."""
+    try:
+        from . import browser_metrics
+        browser_metrics.record_skill_event(kind, host, sig, rev=rev, state=state, extra=extra or None)
+    except Exception:
+        pass
+
+
 # In-memory hot cache: key "host::task_sig" -> skill dict. Bounded.
 _skills: dict[str, dict] = {}
 _MAX_MEM_SKILLS = 200
 _MAX_DISK_SKILLS = 1000          # bound the on-disk library; evict oldest by mtime
 _SKILL_FORMAT_VERSION = 1
 
+# Trust state (the verify gate). A skill moves PROBATION -> TRUSTED only by a
+# successful end-to-end replay; an unproven (probation) skill that fails a replay
+# goes to QUARANTINE and is never replayed again (task falls back to pure LLM).
+_PROBATION = "probation"
+_TRUSTED = "trusted"
+_QUARANTINE = "quarantine"
+# A proven skill tolerates this many consecutive transient replay misses before
+# it's demoted back to probation (forced to re-earn trust).
+_FAIL_DEMOTE_THRESHOLD = 2
+
 # Tools that change page state (worth replaying). Reads/meta are never recorded.
 _PRODUCTIVE = {"BrowserType", "BrowserClickIndex", "BrowserClick", "BrowserPressKey", "BrowserScroll"}
 
@@ -269,6 +303,34 @@ def steps_are_persistable(steps: list[dict]) -> bool:
     return True
 
 
+def _step_key(s: dict) -> tuple:
+    """Canonical identity of a step, ignoring volatile detail, so we can tell a
+    real EDIT (page changed -> different steps) from a transient re-derivation
+    (same steps, the miss was just a timing blip). A slot and a literal are
+    distinct; a parameter's live value is not part of identity."""
+    p = s.get("params", {})
+    tool = s.get("tool")
+    if tool == "BrowserType":
+        if "value_slot" in p:
+            return (tool, p.get("selector"), "slot", p.get("value_slot"))
+        return (tool, p.get("selector"), "text", p.get("text", ""))
+    if tool == "BrowserClickByName":
+        return (tool, p.get("role", ""), p.get("name", ""))
+    if tool == "BrowserClick":
+        return (tool, p.get("selector"))
+    if tool == "BrowserNavigate":
+        return (tool, _sanitize_url(p.get("url", "")))
+    if tool == "BrowserPressKey":
+        return (tool, p.get("key"))
+    if tool == "BrowserScroll":
+        return (tool, p.get("direction"), p.get("amount"))
+    return (tool, json.dumps(p, sort_keys=True, default=str))
+
+
+def _steps_equal(a: list[dict], b: list[dict]) -> bool:
+    return [_step_key(s) for s in a] == [_step_key(s) for s in b]
+
+
 def _sanitized_steps_for_disk(steps: list[dict]) -> list[dict]:
     """Copy of steps safe to persist: navigate URLs stripped of userinfo+fragment."""
     out = []
@@ -320,6 +382,10 @@ def _persist(host: str, sig: str, skill: dict) -> None:
         "steps": _sanitized_steps_for_disk(skill["steps"]),
         "recorded_at": skill.get("recorded_at", time.time()),
         "replays": skill.get("replays", 0),
+        "rev": skill.get("rev", 1),
+        "state": skill.get("state", _PROBATION),
+        "fails": skill.get("fails", 0),
+        "composed_of": skill.get("composed_of", []),
     }
     try:
         d = os.path.dirname(path)
@@ -362,85 +428,19 @@ def _load_from_disk(host: str, sig: str) -> dict | None:
             "host": data.get("host", host), "task_sig": data.get("task_sig", sig),
             "steps": data["steps"], "recorded_at": data.get("recorded_at", 0),
             "replays": data.get("replays", 0), "persisted": True,
+            "rev": data.get("rev", 1), "state": data.get("state", _PROBATION),
+            "fails": data.get("fails", 0), "composed_of": data.get("composed_of", []),
         }
     except Exception as e:
         logger.debug(f"[browser-skills] load failed: {e}")
         return None
 
 
-def record_skill(host: str, task: str, action_log: list[dict]) -> bool:
-    """Record a replayable skill. Non-sensitive skills persist to disk;
-    sensitive ones stay in-memory only. Returns True if a skill was stored
-    (memory or disk). Best-effort; never raises into the caller."""
-    try:
-        if not host:
-            return False
-        steps = distill_steps(action_log)
-        if not steps:
-            return False
-        sig = _sig(task)
-        if not sig:
-            return False
-        steps = _parameterize(steps, task)  # quoted values -> slots (not stored)
-        persistable = steps_are_persistable(steps)
-        skill = {
-            "host": host, "task_sig": sig, "steps": steps,
-            "recorded_at": time.time(), "replays": 0, "persisted": persistable,
-        }
-        _skills[_key(host, sig)] = skill
-        if len(_skills) > _MAX_MEM_SKILLS:
-            oldest = min(_skills, key=lambda k: _skills[k]["recorded_at"])
-            _skills.pop(oldest, None)
-        if persistable:
-            _persist(host, sig, skill)
-            logger.info(f"[browser-skills] recorded + PERSISTED {len(steps)}-step skill for {host}")
-        else:
-            logger.info(f"[browser-skills] recorded {len(steps)}-step skill for {host} (in-memory only: sensitive)")
-        return True
-    except Exception as e:
-        logger.debug(f"[browser-skills] record failed: {e}")
-        return False
-
-
-def find_skill(host: str, task: str) -> dict | None:
-    """Exact-key lookup: in-memory hot cache first, then a single lazy disk read
-    (no corpus scan). Returns the skill or None. Cheap + flat as the library grows."""
-    if not host:
-        return None
-    sig = _sig(task)
-    if not sig:
-        return None
-    k = _key(host, sig)
-    hit = _skills.get(k)
-    if hit:
-        return hit
-    loaded = _load_from_disk(host, sig)
-    if loaded:
-        _skills[k] = loaded  # warm the hot cache
-        return loaded
-    return None
-
-
-def mark_replayed(host: str, task: str) -> None:
-    s = find_skill(host, task)
-    if s:
-        s["replays"] = s.get("replays", 0) + 1
-        if s.get("persisted"):
-            _persist(host, s["task_sig"], s)  # keep the on-disk count fresh
-
-
-def list_skills(host: str) -> list[dict]:
-    """Compact summaries of the skills learned for a host (task + step count +
-    replay count), NOT full step dumps, so the agent can ask "what shortcuts do
-    I have here?" without pulling a wall of detail into context. Reads in-memory
-    + the on-disk library for this host."""
+def _host_skills(host: str) -> dict[str, dict]:
+    """Every skill for one host, keyed by task_sig, in-memory authoritative over
+    disk. One flat scan of the library dir (same cost list_skills always paid);
+    callers that run per-record gate on cheap pre-checks before calling."""
     out: dict[str, dict] = {}
-    for s in _skills.values():
-        if s.get("host") == host:
-            out[s["task_sig"]] = {
-                "task": s["task_sig"], "steps": len(s.get("steps", [])),
-                "replays": s.get("replays", 0), "persisted": s.get("persisted", False),
-            }
     d = _skills_dir()
     if d:
         try:
@@ -452,22 +452,228 @@ def list_skills(host: str) -> list[dict]:
                         data = json.load(fh)
                 except Exception:
                     continue
-                sig = data.get("task_sig")
-                if data.get("host") == host and sig and sig not in out:
-                    out[sig] = {
-                        "task": sig, "steps": len(data.get("steps", [])),
-                        "replays": data.get("replays", 0), "persisted": True,
-                    }
+                if data.get("host") == host and data.get("task_sig"):
+                    out[data["task_sig"]] = {**data, "persisted": True}
         except Exception:
             pass
-    return sorted(out.values(), key=lambda x: -x["replays"])
+    for s in _skills.values():
+        if s.get("host") == host and s.get("task_sig"):
+            out[s["task_sig"]] = s
+    return out
+
+
+# --- composition (build on what's already proven) --------------------------
+# When a freshly learned skill's steps OPEN with the full step list of an
+# already-TRUSTED skill on the same host, we record that it "builds on" the
+# sub-skill. The big steps stay inline (the skill is self-contained and robust on
+# its own); the link is provenance + a safety wire: if that foundation is later
+# deprecated or goes stale, every skill built on it is knocked back to probation
+# so it must re-prove instead of silently riding a now-broken sub-sequence.
+_COMPOSE_MIN_SUB_STEPS = 2
+
+
+def _detect_composition(host: str, sig: str, steps: list[dict]) -> list[str]:
+    """Sigs of TRUSTED host skills whose full step list is a strict opening
+    prefix of `steps`. Gated: needs a tail, so only runs for >=3-step skills."""
+    if len(steps) < _COMPOSE_MIN_SUB_STEPS + 1:
+        return []
+    keys = [_step_key(s) for s in steps]
+    found: list[str] = []
+    for other_sig, other in _host_skills(host).items():
+        if other_sig == sig or other.get("state") != _TRUSTED:
+            continue
+        osteps = other.get("steps", [])
+        if len(osteps) < _COMPOSE_MIN_SUB_STEPS or len(osteps) >= len(steps):
+            continue
+        if [_step_key(s) for s in osteps] == keys[: len(osteps)]:
+            found.append(other_sig)
+    return found
+
+
+def _invalidate_dependents(host: str, sub_sig: str) -> None:
+    """Knock every skill that builds on `sub_sig` back to probation: its proven
+    foundation just moved (edited/deprecated/demoted), so it must re-earn trust
+    rather than ghost-ride a sub-sequence that may no longer hold."""
+    for dep_sig, dep in _host_skills(host).items():
+        if sub_sig in dep.get("composed_of", []) and dep.get("state") == _TRUSTED:
+            k = _key(host, dep_sig)
+            live = _skills.get(k) or dep
+            live["state"] = _PROBATION
+            live["fails"] = 0
+            _skills[k] = live
+            if live.get("persisted"):
+                _persist(host, dep_sig, live)
+            _event("invalidate", host, dep_sig, rev=live.get("rev", 1), state=_PROBATION, foundation=sub_sig)
+            logger.info(f"[browser-skills] {host}::{dep_sig} knocked to probation "
+                        f"(its foundation {sub_sig} changed)")
+
+
+def record_skill(host: str, task: str, action_log: list[dict]) -> bool:
+    """Record (or EDIT) a replayable skill. Non-sensitive skills persist to disk;
+    sensitive ones stay in-memory only. Edit-aware: if a skill already exists for
+    this (host, task) and the freshly distilled steps DIFFER, this is a real edit
+    (the page changed) so we re-version it (rev++) back to probation; if they're
+    IDENTICAL the prior replay miss was transient, so we keep the existing trust
+    and just clear the fail streak. Returns True if a skill is in place after the
+    call. Best-effort; never raises into the caller."""
+    try:
+        if not host:
+            return False
+        steps = distill_steps(action_log)
+        if not steps:
+            return False
+        sig = _sig(task)
+        if not sig:
+            return False
+        steps = _parameterize(steps, task)  # quoted values -> slots (not stored)
+        persistable = steps_are_persistable(steps)
+        k = _key(host, sig)
+        existing = _skills.get(k) or _load_from_disk(host, sig)
+
+        if existing and _steps_equal(existing.get("steps", []), steps):
+            # Same skill re-derived: the replay that triggered this was a transient
+            # miss, not a stale skill. Keep rev + trust; just clear the fail streak.
+            # If it was quarantined (a known-bad distillation), leave it quarantined
+            # so the task keeps running on the pure-LLM baseline, never re-replayed.
+            existing["fails"] = 0
+            existing["recorded_at"] = time.time()
+            existing["persisted"] = persistable
+            _skills[k] = existing
+            if persistable:
+                _persist(host, sig, existing)
+            logger.info(f"[browser-skills] re-derived identical {len(steps)}-step skill for {host} "
+                        f"(rev {existing.get('rev', 1)}, state={existing.get('state')}, transient miss)")
+            return True
+
+        rev = (existing.get("rev", 1) + 1) if existing else 1
+        skill = {
+            "host": host, "task_sig": sig, "steps": steps,
+            "recorded_at": time.time(), "replays": 0, "persisted": persistable,
+            "rev": rev, "state": _PROBATION, "fails": 0,
+            "composed_of": _detect_composition(host, sig, steps),
+        }
+        _skills[k] = skill
+        if len(_skills) > _MAX_MEM_SKILLS:
+            oldest = min(_skills, key=lambda kk: _skills[kk]["recorded_at"])
+            _skills.pop(oldest, None)
+        if persistable:
+            _persist(host, sig, skill)
+        verb = "EDITED" if existing else "learned"
+        comp = f", builds on {skill['composed_of']}" if skill["composed_of"] else ""
+        logger.info(f"[browser-skills] {verb} {len(steps)}-step skill for {host} "
+                    f"(rev {rev}, probationary{', persisted' if persistable else ', in-memory only: sensitive'}{comp})")
+        _event("edit" if existing else "learn", host, sig, rev=rev, state=_PROBATION,
+               steps=len(steps), composed_of=skill["composed_of"], persisted=persistable)
+        if skill["composed_of"]:
+            _event("compose", host, sig, rev=rev, state=_PROBATION, builds_on=skill["composed_of"])
+        if existing:
+            _invalidate_dependents(host, sig)  # anything built on the OLD version must re-prove
+        return True
+    except Exception as e:
+        logger.debug(f"[browser-skills] record failed: {e}")
+        return False
+
+
+def find_skill(host: str, task: str) -> dict | None:
+    """Exact-key lookup for REPLAY: in-memory hot cache first, then a single lazy
+    disk read (no corpus scan). A QUARANTINED skill (unproven and already failed)
+    is never handed back, so the task runs on the pure-LLM baseline instead of
+    re-attempting a known-bad replay. Cheap + flat as the library grows."""
+    if not host:
+        return None
+    sig = _sig(task)
+    if not sig:
+        return None
+    k = _key(host, sig)
+    hit = _skills.get(k)
+    if not hit:
+        loaded = _load_from_disk(host, sig)
+        if loaded:
+            _skills[k] = loaded  # warm the hot cache (even if quarantined)
+            hit = loaded
+    if not hit or hit.get("state") == _QUARANTINE:
+        return None
+    return hit
+
+
+def mark_replay_succeeded(host: str, task: str) -> None:
+    """A replay ran end to end. Count it and, if the skill was still on
+    probation, PROMOTE it to trusted (the verify gate just passed)."""
+    s = find_skill(host, task)
+    if not s:
+        return
+    s["replays"] = s.get("replays", 0) + 1
+    s["fails"] = 0
+    promoted = s.get("state") != _TRUSTED
+    s["state"] = _TRUSTED
+    if s.get("persisted"):
+        _persist(host, s["task_sig"], s)  # keep the on-disk count + state fresh
+    if promoted:
+        logger.info(f"[browser-skills] {host}::{s['task_sig']} PROVEN by replay (rev {s.get('rev', 1)}) -> trusted")
+        _event("promote", host, s["task_sig"], rev=s.get("rev", 1), state=_TRUSTED, replays=s["replays"])
+
+
+def mark_replay_failed(host: str, task: str) -> str:
+    """A replay failed mid-way. Update trust and report what happened so the
+    caller can log it; the caller then falls through to the full LLM agent (which
+    re-records, edit-aware). Returns one of:
+      'quarantined' - skill was unproven (probation) and failed -> never replay it
+                      again; the task runs on the pure-LLM baseline from now on.
+      'demoted'     - a trusted skill crossed the transient-miss threshold -> back
+                      to probation (must re-earn trust).
+      'kept'        - a trusted skill's first transient miss; left in place.
+      'none'        - no live (non-quarantined) skill for this task."""
+    s = find_skill(host, task)
+    if not s:
+        return "none"
+    sig = s["task_sig"]
+    if s.get("state") != _TRUSTED:
+        s["state"] = _QUARANTINE
+        s["fails"] = s.get("fails", 0) + 1
+        if s.get("persisted"):
+            _persist(host, sig, s)
+        _event("quarantine", host, sig, rev=s.get("rev", 1), state=_QUARANTINE)
+        _invalidate_dependents(host, sig)
+        logger.info(f"[browser-skills] {host}::{sig} (unproven) failed replay -> quarantined (baseline from here)")
+        return "quarantined"
+    s["fails"] = s.get("fails", 0) + 1
+    if s["fails"] >= _FAIL_DEMOTE_THRESHOLD:
+        s["state"] = _PROBATION
+        if s.get("persisted"):
+            _persist(host, sig, s)
+        _event("demote", host, sig, rev=s.get("rev", 1), state=_PROBATION, fails=s["fails"])
+        _invalidate_dependents(host, sig)
+        logger.info(f"[browser-skills] {host}::{sig} failed {s['fails']}x -> demoted to probation")
+        return "demoted"
+    if s.get("persisted"):
+        _persist(host, sig, s)
+    logger.info(f"[browser-skills] {host}::{sig} transient replay miss ({s['fails']}/{_FAIL_DEMOTE_THRESHOLD}), trust kept")
+    return "kept"
+
+
+def list_skills(host: str) -> list[dict]:
+    """Compact summaries of the skills learned for a host (task + step count +
+    replay count + trust state + what it builds on), NOT full step dumps, so the
+    agent can ask "what shortcuts do I have here?" without pulling a wall of
+    detail into context. Reads in-memory + the on-disk library for this host."""
+    out = []
+    for sig, s in _host_skills(host).items():
+        out.append({
+            "task": sig, "steps": len(s.get("steps", [])),
+            "replays": s.get("replays", 0), "persisted": s.get("persisted", False),
+            "state": s.get("state", _PROBATION), "rev": s.get("rev", 1),
+            "builds_on": list(s.get("composed_of", [])),
+        })
+    # trusted first, then most-reused
+    return sorted(out, key=lambda x: (x["state"] != _TRUSTED, -x["replays"]))
 
 
 def deprecate_skill(host: str, task: str) -> bool:
-    """Remove a skill (in-memory + disk) so it stops being replayed. The agent
-    calls this when it judges a saved shortcut is stale / wrong (page changed).
-    Accepts either the raw task or the task_sig from list_skills (sig is
-    idempotent under _sig). Returns True if something was removed."""
+    """Remove a skill (in-memory + disk) so it stops being replayed, and knock
+    any skill that was built on it back to probation. The agent calls this when
+    it judges a saved shortcut is stale / wrong (page changed). Accepts either the
+    raw task or the task_sig from list_skills (sig is idempotent under _sig).
+    Returns True if something was removed."""
     if not host:
         return False
     sig = _sig(task)
@@ -480,6 +686,7 @@ def deprecate_skill(host: str, task: str) -> bool:
         except Exception:
             pass
     if removed:
+        _invalidate_dependents(host, sig)
         logger.info(f"[browser-skills] deprecated skill {host}::{sig}")
     return removed
 

backend/tests/test_browser_agent_loop.py

@@ -268,6 +268,70 @@ def test_replay_falls_back_to_full_agent_when_a_step_fails(monkeypatch):
     assert len(primary.calls) > 0, "fell back to the full LLM agent"
 
 
+def test_replay_success_promotes_skill_to_trusted_through_the_loop(monkeypatch):
+    # The verify gate, end to end: run 1 learns a PROBATION skill; run 2 replays
+    # it successfully, which must PROMOTE it to trusted (proven by a real replay).
+    import backend.apps.agents.browser.browser_skills as SK
+    SK.clear()
+    BH._browser_history.clear(); BH._domain_notes.clear()
+    primary = FakeLLM([
+        Resp([_rp("click submit"), _tu("BrowserListInteractives")]),
+        Resp([_rp("click it"), _tu("BrowserClickIndex", index=1)]),
+        Resp([Blk("text", "Done.")], stop_reason="end_turn"),
+    ])
+    aux = FakeAux()
+    _install(monkeypatch, primary, aux)
+    asyncio.run(BA.run_browser_agent(
+        task="click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
+    ))
+    assert SK.find_skill("docs.google.com", "click the Submit button")["state"] == SK._PROBATION
+    r2 = asyncio.run(BA.run_browser_agent(
+        task="click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
+    ))
+    assert r2.get("replayed") is True
+    assert SK.find_skill("docs.google.com", "click the Submit button")["state"] == SK._TRUSTED
+
+
+def test_unproven_skill_that_fails_is_quarantined_and_never_retried(monkeypatch):
+    # The anti-ghost guard, end to end: an unproven skill that fails a replay must
+    # be quarantined so the NEXT run does not even attempt the (known-bad) replay,
+    # it goes straight to the pure-LLM baseline. A silent re-fail would be a ghost.
+    import backend.apps.agents.browser.browser_skills as SK
+    SK.clear()
+    BH._browser_history.clear()
+    SK.record_skill("docs.google.com", "click the Save button", [
+        {"tool": "BrowserClickIndex", "input": {"index": 1}, "ok": True,
+         "clicked_role": "button", "clicked_name": "Save"},
+    ])  # probation, unproven
+    primary = FakeLLM([Resp([Blk("text", "full agent handled it")], stop_reason="end_turn")])
+    aux = FakeAux()
+    sent = _install(monkeypatch, primary, aux)
+    orig = BA.ws_manager.send_browser_command
+
+    async def _fail_cbn(request_id, action, browser_id, params, tab_id=""):
+        if action == "click_by_name":
+            sent.append({"action": action, "params": params})
+            return {"error": 'No element matching name="Save" on this page.'}
+        return await orig(request_id, action, browser_id, params, tab_id)
+    monkeypatch.setattr(BA.ws_manager, "send_browser_command", _fail_cbn, raising=False)
+
+    # Run 1: replay is attempted, the step fails -> skill is quarantined.
+    asyncio.run(BA.run_browser_agent(
+        task="click the Save button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
+    ))
+    assert any(c["action"] == "click_by_name" for c in sent), "run 1 DID attempt the replay"
+    assert SK.list_skills("docs.google.com")[0]["state"] == SK._QUARANTINE
+
+    # Run 2: the quarantined skill must NOT be replayed again.
+    sent.clear()
+    r2 = asyncio.run(BA.run_browser_agent(
+        task="click the Save button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
+    ))
+    assert not r2.get("replayed")
+    assert not any(c["action"] == "click_by_name" for c in sent), \
+        "a quarantined skill must never be replayed again (would be a ghost re-fail)"
+
+
 def test_perception_is_frontloaded_into_first_turn(monkeypatch):
     # With a known start URL, the agent should prefetch the element list + page
     # text and put them in the FIRST user message, so the model can act on turn 1

backend/tests/test_browser_skills.py

@@ -273,3 +273,188 @@ def test_deprecate_removes_skill_from_memory_and_disk(_isolated_skills):
 
 def test_deprecate_unknown_is_false(_isolated_skills):
     assert sk.deprecate_skill("shop.com", "never recorded this") is False
+
+
+# --- versioned safe-edit: the trust gate ----------------------------------
+# A skill is never trusted until a real replay proves it; an unproven skill that
+# fails is quarantined (never replayed again) so a lossy skill can't ghost-succeed
+# or run slower-than-baseline; re-deriving different steps is a re-versioned EDIT.
+
+def test_new_skill_starts_on_probation(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["state"] == sk._PROBATION and s["rev"] == 1 and s["replays"] == 0
+
+
+def test_replay_success_promotes_probation_to_trusted(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["state"] == sk._TRUSTED and s["replays"] == 1 and s["fails"] == 0
+
+
+def test_probation_failure_quarantines_and_blocks_future_replay(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())          # probation
+    verdict = sk.mark_replay_failed("shop.com", "do a thing now")
+    assert verdict == "quarantined"
+    # the ghost guard: a quarantined skill is NEVER handed back for replay...
+    assert sk.find_skill("shop.com", "do a thing now") is None
+    # ...but the record still exists (visible + deprecatable), it just won't run
+    listed = sk.list_skills("shop.com")
+    assert len(listed) == 1 and listed[0]["state"] == sk._QUARANTINE
+
+
+def test_quarantined_skill_re_recorded_identical_stays_quarantined(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_failed("shop.com", "do a thing now")            # quarantined
+    # the full LLM agent re-runs and distills the SAME (still-lossy) steps:
+    sk.record_skill("shop.com", "do a thing now", _log())
+    # it must stay quarantined -> pure-LLM baseline, never a wasted replay again
+    assert sk.find_skill("shop.com", "do a thing now") is None
+    assert sk.list_skills("shop.com")[0]["state"] == sk._QUARANTINE
+
+
+def test_quarantined_skill_unquarantines_on_a_real_edit(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_failed("shop.com", "do a thing now")            # quarantined
+    # now the page changed and the LLM derives a DIFFERENT click -> a real edit,
+    # which earns the skill another chance (back on probation, re-versioned)
+    edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
+                             "clicked_role": "button", "clicked_name": "Submit"}]
+    sk.record_skill("shop.com", "do a thing now", edited)
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s is not None and s["state"] == sk._PROBATION and s["rev"] == 2
+
+
+def test_trusted_skill_tolerates_one_transient_miss_then_demotes(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")         # trusted
+    assert sk.mark_replay_failed("shop.com", "do a thing now") == "kept"
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["state"] == sk._TRUSTED and s["fails"] == 1           # still usable
+    assert sk.mark_replay_failed("shop.com", "do a thing now") == "demoted"
+    assert sk.find_skill("shop.com", "do a thing now")["state"] == sk._PROBATION
+
+
+def test_re_record_identical_keeps_trust_and_rev(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")
+    sk.find_skill("shop.com", "do a thing now")["replays"] = 5     # pretend reused a lot
+    sk.record_skill("shop.com", "do a thing now", _log())          # identical re-derive
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["state"] == sk._TRUSTED and s["rev"] == 1 and s["replays"] == 5
+
+
+def test_re_record_different_is_an_edit_that_reversions_to_probation(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")         # trusted, rev 1
+    edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
+                             "clicked_role": "button", "clicked_name": "Submit"}]
+    sk.record_skill("shop.com", "do a thing now", edited)          # different -> EDIT
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["rev"] == 2 and s["state"] == sk._PROBATION and s["replays"] == 0
+    cbn = next(x for x in s["steps"] if x["tool"] == "BrowserClickByName")
+    assert cbn["params"]["name"] == "Submit"                       # the new step stuck
+
+
+def test_rev_and_state_persist_across_restart(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")
+    edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
+                             "clicked_role": "button", "clicked_name": "Submit"}]
+    sk.record_skill("shop.com", "do a thing now", edited)          # rev 2, probation
+    sk.clear(wipe_disk=False)                                      # restart
+    s = sk.find_skill("shop.com", "do a thing now")
+    assert s["rev"] == 2 and s["state"] == sk._PROBATION
+
+
+def test_steps_equal_distinguishes_slot_from_literal_and_changed_click():
+    nav = {"tool": "BrowserNavigate", "params": {"url": "https://x.com/a#frag"}}
+    nav2 = {"tool": "BrowserNavigate", "params": {"url": "https://x.com/a"}}   # frag stripped == same
+    lit = {"tool": "BrowserType", "params": {"selector": "#q", "text": "shoes"}}
+    slot = {"tool": "BrowserType", "params": {"selector": "#q", "value_slot": 0}}
+    send = {"tool": "BrowserClickByName", "params": {"role": "button", "name": "Send"}}
+    submit = {"tool": "BrowserClickByName", "params": {"role": "button", "name": "Submit"}}
+    assert sk._steps_equal([nav], [nav2])           # fragment-only diff is NOT an edit
+    assert not sk._steps_equal([lit], [slot])       # literal vs parameterized IS an edit
+    assert not sk._steps_equal([send], [submit])    # renamed button IS an edit
+
+
+def test_mark_replay_helpers_on_unknown_are_safe(_isolated_skills):
+    sk.mark_replay_succeeded("shop.com", "never recorded")    # no raise
+    assert sk.mark_replay_failed("shop.com", "never recorded") == "none"
+
+
+def test_demoted_skill_can_be_re_proven(_isolated_skills):
+    sk.record_skill("shop.com", "do a thing now", _log())
+    sk.mark_replay_succeeded("shop.com", "do a thing now")    # trusted
+    sk.mark_replay_failed("shop.com", "do a thing now")
+    sk.mark_replay_failed("shop.com", "do a thing now")       # demoted to probation
+    assert sk.find_skill("shop.com", "do a thing now")["state"] == sk._PROBATION
+    sk.mark_replay_succeeded("shop.com", "do a thing now")    # earns trust back
+    assert sk.find_skill("shop.com", "do a thing now")["state"] == sk._TRUSTED
+
+
+# --- composition: build on what's already proven, propagate staleness -------
+
+def _log_plus():
+    # distills to _log()'s 3 steps PLUS a 4th click -> a strict superset sequence
+    return _log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
+                      "clicked_role": "button", "clicked_name": "Checkout"}]
+
+
+def _trust(host, task, log):
+    sk.record_skill(host, task, log)
+    sk.mark_replay_succeeded(host, task)
+
+
+def test_composition_links_to_trusted_sub_skill(_isolated_skills):
+    _trust("shop.com", "search shoes now", _log())                # trusted foundation
+    sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
+    c = sk.find_skill("shop.com", "search shoes and checkout now")
+    assert c["composed_of"] == [sk._sig("search shoes now")]
+
+
+def test_composition_ignores_untrusted_foundation(_isolated_skills):
+    sk.record_skill("shop.com", "search shoes now", _log())        # probation, NOT trusted
+    sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
+    c = sk.find_skill("shop.com", "search shoes and checkout now")
+    assert c["composed_of"] == []          # only a PROVEN sub-skill is built upon
+
+
+def test_deprecating_a_foundation_demotes_everything_built_on_it(_isolated_skills):
+    _trust("shop.com", "search shoes now", _log())
+    _trust("shop.com", "search shoes and checkout now", _log_plus())   # composed + trusted
+    assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._TRUSTED
+    sk.deprecate_skill("shop.com", "search shoes now")            # foundation pulled
+    # the ghost guard for composition: the dependent must NOT stay trusted on a
+    # foundation that no longer exists; it's knocked back to re-prove
+    assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
+
+
+def test_demoting_a_foundation_demotes_its_dependents(_isolated_skills):
+    _trust("shop.com", "search shoes now", _log())
+    _trust("shop.com", "search shoes and checkout now", _log_plus())
+    sk.mark_replay_failed("shop.com", "search shoes now")
+    sk.mark_replay_failed("shop.com", "search shoes now")         # foundation demoted
+    assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
+
+
+def test_editing_a_foundation_demotes_its_dependents(_isolated_skills):
+    _trust("shop.com", "search shoes now", _log())
+    _trust("shop.com", "search shoes and checkout now", _log_plus())
+    edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
+                             "clicked_role": "button", "clicked_name": "Find"}]
+    sk.record_skill("shop.com", "search shoes now", edited)       # foundation changed
+    assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
+
+
+def test_list_skills_surfaces_state_rev_and_builds_on(_isolated_skills):
+    _trust("shop.com", "search shoes now", _log())
+    sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
+    listed = {x["task"]: x for x in sk.list_skills("shop.com")}
+    foundation = listed[sk._sig("search shoes now")]
+    composed = listed[sk._sig("search shoes and checkout now")]
+    assert foundation["state"] == sk._TRUSTED and foundation["builds_on"] == []
+    assert composed["builds_on"] == [sk._sig("search shoes now")]
+    assert "rev" in composed and "steps" in composed