#!/usr/bin/env python3
"""
Genesis Browser Skill Framework
================================
Pre-mapped browser workflows stored as skills, invoked by any agent via
Gemini Flash agentic vision. Skills are JSON workflow maps loaded from
.claude/skills/browser/, executed with Playwright, and stored to Graphiti
for RLM (Recursive Language Model) memory recall.

Architecture:
    Skill JSON  →  BrowserSkillEngine.execute_skill()
                       │
                       ├─ Playwright (local headless or Browserless)
                       │
                       ├─ Gemini 3 Flash vision (Think→Act→Observe loop)
                       │
                       └─ Graphiti episode store (success) / skill map fix (failure)

Memory Model (Ebbinghaus-inspired):
    - Skills not used in 30 days get a re-verify flag set
    - Every successful run stores a Graphiti episode with screenshot hashes
    - Next run recalls from Graphiti FIRST, uses static JSON as fallback

Usage:
    python3 core/browser_skill_engine.py --list
    python3 core/browser_skill_engine.py --list --platform gohighlevel
    python3 core/browser_skill_engine.py --execute ghl_create_subaccount_api_integration \
        --params '{"subaccount_name": "Test Co", "integration_name": "Genesis API"}'
    python3 core/browser_skill_engine.py --learn gohighlevel "create a new pipeline"

MCP Tool Registration (genesis-core server):
    execute_browser_skill(skill_id, params)
    list_available_skills(platform)
    learn_browser_skill(platform, goal)

Author: Genesis System (Session 59)
Date: 2026-02-24
"""

import argparse
import asyncio
import base64
import hashlib
import json
import logging
import os
import sys
import time
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any, Optional

# ── Path bootstrap ─────────────────────────────────────────────────────────────
GENESIS_ROOT = Path("/mnt/e/genesis-system")
sys.path.insert(0, str(GENESIS_ROOT))

# ── Secrets loader ─────────────────────────────────────────────────────────────
def _load_secrets():
    """Load config/secrets.env into os.environ (idempotent)."""
    secrets_path = GENESIS_ROOT / "config" / "secrets.env"
    if not secrets_path.exists():
        return
    for raw in secrets_path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        key = key.strip()
        val = val.strip().strip('"').strip("'")
        if key and val:
            os.environ.setdefault(key, val)

_load_secrets()

# ── Directory layout ───────────────────────────────────────────────────────────
SKILLS_DIR      = GENESIS_ROOT / ".claude" / "skills" / "browser"
SCREENSHOT_DIR  = GENESIS_ROOT / "data" / "screenshots" / "browser_skills"
LOG_DIR         = GENESIS_ROOT / "logs"
SKILL_STATE_DB  = GENESIS_ROOT / "data" / "browser_skill_state.json"

for _d in [SKILLS_DIR, SCREENSHOT_DIR, LOG_DIR, SKILL_STATE_DB.parent]:
    _d.mkdir(parents=True, exist_ok=True)

# ── Logging ────────────────────────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [BSE] %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler(LOG_DIR / "browser_skill_engine.log", encoding="utf-8"),
        logging.StreamHandler(sys.stdout),
    ],
)
log = logging.getLogger("browser_skill_engine")

# ── Gemini setup ───────────────────────────────────────────────────────────────
GEMINI_API_KEY = (
    os.environ.get("GEMINI_API_KEY_NEW")
    or os.environ.get("GEMINI_API_KEY")
    or ""
)
VISION_MODEL = "gemini-2.0-flash"   # fallback if gemini-3-flash-preview unavailable
PREFERRED_VISION_MODEL = "gemini-3-flash-preview"

try:
    from google import genai as google_genai
    from google.genai import types as genai_types
    GEMINI_AVAILABLE = bool(GEMINI_API_KEY)
    if GEMINI_AVAILABLE:
        _gemini_client = google_genai.Client(api_key=GEMINI_API_KEY)
    else:
        _gemini_client = None
except ImportError:
    GEMINI_AVAILABLE = False
    _gemini_client = None
    log.warning("google-genai not installed — vision guidance disabled. pip install google-genai")

# ── Playwright setup ───────────────────────────────────────────────────────────
BROWSERLESS_URL = os.environ.get("BROWSERLESS_URL", "")
BROWSERLESS_TOKEN = os.environ.get("BROWSERLESS_TOKEN", "")
PLAYWRIGHT_TIMEOUT = int(os.environ.get("PLAYWRIGHT_TIMEOUT_MS", "30000"))

try:
    from playwright.async_api import async_playwright, Page, Browser, BrowserContext
    PLAYWRIGHT_AVAILABLE = True
except ImportError:
    PLAYWRIGHT_AVAILABLE = False
    log.warning("playwright not installed — browser execution disabled. pip install playwright")

# ── Graphiti / RLM setup ───────────────────────────────────────────────────────
GRAPHITI_MCP_URL = os.environ.get("GRAPHITI_MCP_URL", "http://152.53.201.221:8001/mcp")

try:
    import requests as _req_lib
    REQUESTS_AVAILABLE = True
except ImportError:
    REQUESTS_AVAILABLE = False


# =============================================================================
# Skill loader helpers
# =============================================================================

def _load_skill(skill_id: str) -> dict:
    """Load a skill definition from the skills directory by ID."""
    skill_file = SKILLS_DIR / f"{skill_id}.json"
    if not skill_file.exists():
        raise FileNotFoundError(f"Skill '{skill_id}' not found at {skill_file}")
    with open(skill_file, encoding="utf-8") as f:
        return json.load(f)


def _save_skill(skill: dict) -> None:
    """Persist an updated skill definition back to disk."""
    skill_id = skill["skill_id"]
    skill_file = SKILLS_DIR / f"{skill_id}.json"
    with open(skill_file, "w", encoding="utf-8") as f:
        json.dump(skill, f, indent=2, ensure_ascii=False)
    log.info("Skill '%s' updated on disk", skill_id)


def _list_skills(platform: Optional[str] = None) -> list[dict]:
    """Return all skill definitions, optionally filtered by platform."""
    skills = []
    for p in sorted(SKILLS_DIR.glob("*.json")):
        try:
            s = json.loads(p.read_text(encoding="utf-8"))
            if platform is None or s.get("platform", "").lower() == platform.lower():
                skills.append(s)
        except json.JSONDecodeError as e:
            log.warning("Corrupt skill file %s: %s", p.name, e)
    return skills


def _load_skill_state() -> dict:
    """Load mutable run-state overlay (execution counts, timestamps, etc.)."""
    if SKILL_STATE_DB.exists():
        try:
            return json.loads(SKILL_STATE_DB.read_text(encoding="utf-8"))
        except Exception:
            pass
    return {}


def _save_skill_state(state: dict) -> None:
    """Persist run-state overlay."""
    with open(SKILL_STATE_DB, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2, ensure_ascii=False)


def _resolve_params(text: str, params: dict, env_vars: list[str]) -> str:
    """Substitute {param} and {ENV_VAR} placeholders in instruction text."""
    result = text
    for key, val in params.items():
        result = result.replace(f"{{{key}}}", str(val))
    for var in env_vars:
        env_val = os.environ.get(var, "")
        result = result.replace(f"{{{var}}}", env_val)
    return result


# =============================================================================
# Graphiti RLM integration
# =============================================================================

class GraphitiClient:
    """Lightweight Graphiti MCP client for episode storage and recall."""

    def __init__(self, url: str = GRAPHITI_MCP_URL):
        self.url = url
        self._available: Optional[bool] = None

    def _check_available(self) -> bool:
        if not REQUESTS_AVAILABLE:
            return False
        if self._available is None:
            try:
                r = _req_lib.get(self.url.replace("/mcp", "/health"), timeout=3)
                self._available = r.status_code < 500
            except Exception:
                self._available = False
        return self._available

    def store_episode(self, name: str, body: str, source: str = "browser_skill") -> dict:
        """Store a browser skill execution episode in Graphiti."""
        if not self._check_available():
            log.warning("Graphiti unavailable — episode '%s' not stored", name)
            return {"stored": False, "reason": "graphiti_unavailable"}

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "add_memory",
                "arguments": {
                    "name": name,
                    "episode_body": body,
                    "source": source,
                    "source_description": f"Browser skill execution: {name}",
                }
            }
        }
        try:
            r = _req_lib.post(self.url, json=payload, timeout=15)
            result = r.json()
            episode_uuid = result.get("result", {}).get("episode_uuid", "unknown")
            log.info("Graphiti episode stored: %s (uuid=%s)", name, episode_uuid)
            return {"stored": True, "episode_uuid": episode_uuid}
        except Exception as e:
            log.warning("Graphiti store failed: %s", e)
            return {"stored": False, "reason": str(e)}

    def search_episodes(self, query: str, limit: int = 5) -> list[dict]:
        """Search Graphiti for relevant browser skill episodes."""
        if not self._check_available():
            return []
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "search_memory",
                "arguments": {"query": query, "num_results": limit}
            }
        }
        try:
            r = _req_lib.post(self.url, json=payload, timeout=10)
            facts = r.json().get("result", {}).get("facts", [])
            return facts
        except Exception as e:
            log.warning("Graphiti search failed: %s", e)
            return []


_graphiti = GraphitiClient()


# =============================================================================
# Gemini vision guidance
# =============================================================================

async def _gemini_decide_action(
    screenshot_b64: str,
    instruction: str,
    history: list[str],
    current_url: str,
    step: int,
) -> dict:
    """
    Ask Gemini Flash vision to decide the next browser action.

    Returns a dict with keys:
        action_type: click | type | scroll | navigate | extract | done | fail
        selector: CSS selector or None
        coordinates: [x, y] or None
        text: text to type or None
        url: URL to navigate to or None
        value: extracted value or None
        reasoning: brief explanation
    """
    if not GEMINI_AVAILABLE or _gemini_client is None:
        return {
            "action_type": "fail",
            "reasoning": "Gemini not available — set GEMINI_API_KEY",
        }

    history_text = "\n".join(history[-8:]) if history else "None"
    prompt = f"""You are controlling a web browser to complete a task step.

CURRENT INSTRUCTION: {instruction}
CURRENT URL: {current_url}
STEP NUMBER: {step}
RECENT ACTIONS (last 8):
{history_text}

Look at the screenshot carefully and decide the SINGLE best next action.

Respond ONLY with valid JSON matching this schema:
{{
  "action_type": "click|type|scroll|navigate|extract|done|fail",
  "selector": "CSS selector string or null",
  "coordinates": [x_int, y_int] or null,
  "text": "text to type if action_type is type, else null",
  "url": "URL if action_type is navigate, else null",
  "value": "extracted text/value if action_type is extract, else null",
  "reasoning": "brief 1-sentence explanation of why this action"
}}

Rules:
- Use "done" when the instruction has been completed successfully
- Use "fail" only if the page state makes it impossible to complete
- Prefer CSS selectors over coordinates when identifiable
- For type actions, always click the field first then type
- If you see a login form and credentials are needed, use action_type "type" for email first
"""

    image_part = genai_types.Part.from_bytes(
        data=base64.b64decode(screenshot_b64),
        mime_type="image/png",
    )

    # Try preferred model first, fall back to stable
    for model_name in [PREFERRED_VISION_MODEL, VISION_MODEL, "gemini-2.0-flash-exp"]:
        try:
            response = _gemini_client.models.generate_content(
                model=model_name,
                contents=[image_part, prompt],
            )
            raw = response.text.strip()
            # Strip markdown code fences if present
            if raw.startswith("```"):
                raw = raw.split("```")[1]
                if raw.startswith("json"):
                    raw = raw[4:]
            return json.loads(raw.strip())
        except json.JSONDecodeError:
            log.warning("Gemini returned non-JSON for step %d — retrying", step)
            continue
        except Exception as e:
            err_str = str(e)
            if "not found" in err_str.lower() or "invalid" in err_str.lower():
                log.warning("Model %s not available, trying next...", model_name)
                continue
            log.error("Gemini vision error (step %d): %s", step, e)
            break

    return {
        "action_type": "fail",
        "reasoning": "Gemini failed to produce valid action JSON after all attempts",
    }


# =============================================================================
# Playwright executor
# =============================================================================

async def _take_screenshot_b64(page) -> str:
    """Capture a full-page screenshot and return as base64 string."""
    raw = await page.screenshot(full_page=False, type="png")
    h = hashlib.sha256(raw).hexdigest()[:12]
    fname = SCREENSHOT_DIR / f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{h}.png"
    fname.write_bytes(raw)
    return base64.b64encode(raw).decode("utf-8"), h


async def _execute_step_action(page, action: dict, step_num: int) -> tuple[bool, str]:
    """
    Execute a single Gemini-decided action on the Playwright page.
    Returns (success, description).
    """
    atype = action.get("action_type", "fail")

    try:
        if atype == "done":
            return True, "Step completed"

        elif atype == "fail":
            return False, f"Gemini declared fail: {action.get('reasoning', '')}"

        elif atype == "navigate":
            url = action.get("url")
            if not url:
                return False, "navigate action missing url"
            await page.goto(url, wait_until="domcontentloaded", timeout=PLAYWRIGHT_TIMEOUT)
            return True, f"Navigated to {url}"

        elif atype == "click":
            sel = action.get("selector")
            coords = action.get("coordinates")
            if sel:
                await page.click(sel, timeout=10000)
                return True, f"Clicked selector: {sel}"
            elif coords:
                await page.mouse.click(coords[0], coords[1])
                return True, f"Clicked coordinates: {coords}"
            else:
                return False, "click action missing selector or coordinates"

        elif atype == "type":
            text = action.get("text", "")
            sel = action.get("selector")
            if sel:
                await page.fill(sel, text, timeout=10000)
            else:
                await page.keyboard.type(text)
            return True, f"Typed text (length={len(text)})"

        elif atype == "scroll":
            coords = action.get("coordinates", [0, 300])
            await page.mouse.wheel(0, coords[1] if isinstance(coords, list) else 300)
            return True, "Scrolled page"

        elif atype == "extract":
            sel = action.get("selector")
            val = action.get("value", "")
            if sel:
                try:
                    el = await page.query_selector(sel)
                    if el:
                        val = await el.inner_text() or await el.get_attribute("value") or val
                except Exception:
                    pass
            return True, f"Extracted: {val[:200]}"

        else:
            return False, f"Unknown action_type: {atype}"

    except Exception as e:
        return False, f"Action execution error: {e}"


async def _run_vision_step(
    page,
    instruction: str,
    env_vars: list,
    params: dict,
    history: list,
    step_num: int,
    max_retries: int = 3,
) -> tuple[bool, str, list]:
    """
    Execute a single vision_guided step with retries.
    Returns (success, description, updated_history).
    """
    resolved = _resolve_params(instruction, params, env_vars)
    current_url = page.url

    for attempt in range(max_retries):
        screenshot_b64, screenshot_hash = await _take_screenshot_b64(page)
        action = await _gemini_decide_action(
            screenshot_b64, resolved, history, current_url, step_num
        )

        log.info(
            "Step %d (attempt %d): %s — %s",
            step_num, attempt + 1,
            action.get("action_type"),
            action.get("reasoning", "")[:80],
        )

        if action.get("action_type") == "done":
            history.append(f"Step {step_num}: DONE — {resolved}")
            return True, "Instruction completed", history

        success, desc = await _execute_step_action(page, action, step_num)
        history.append(
            f"Step {step_num} attempt {attempt+1}: {action.get('action_type')} — {desc}"
        )

        if success:
            await asyncio.sleep(0.8)  # brief settle after action
            # Check if we need more actions for this step
            screenshot_b64_2, _ = await _take_screenshot_b64(page)
            verify_action = await _gemini_decide_action(
                screenshot_b64_2, resolved, history, page.url, step_num
            )
            if verify_action.get("action_type") == "done":
                return True, desc, history
            # Still work to do on this step — let next retry handle it
            continue

        if attempt == max_retries - 1:
            return False, desc, history

        await asyncio.sleep(1.5)

    return False, "Max retries exceeded", history


async def _setup_browser(playwright) -> tuple[Browser, BrowserContext, Page]:
    """Launch Playwright browser, preferring Browserless if configured."""
    if BROWSERLESS_URL and BROWSERLESS_TOKEN:
        log.info("Connecting to Browserless at %s", BROWSERLESS_URL)
        browser = await playwright.chromium.connect_over_cdp(
            f"{BROWSERLESS_URL}?token={BROWSERLESS_TOKEN}"
        )
    else:
        log.info("Launching local headless Chromium")
        browser = await playwright.chromium.launch(
            headless=True,
            args=[
                "--no-sandbox",
                "--disable-blink-features=AutomationControlled",
                "--disable-web-security",
            ],
        )

    context = await browser.new_context(
        viewport={"width": 1440, "height": 900},
        user_agent=(
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/121.0.0.0 Safari/537.36"
        ),
        locale="en-AU",
        timezone_id="Australia/Brisbane",
    )

    # Anti-detection: hide navigator.webdriver
    await context.add_init_script(
        "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
    )

    page = await context.new_page()
    return browser, context, page


# =============================================================================
# Core BrowserSkillEngine class
# =============================================================================

class BrowserSkillEngine:
    """
    Genesis Browser Skill Framework — core execution engine.

    Loads pre-mapped skill JSON workflows, executes them with Playwright
    and Gemini Flash vision guidance, stores successful runs to Graphiti,
    and self-updates skill maps on failure.
    """

    def __init__(self):
        self.skills_dir = SKILLS_DIR
        self.graphiti = _graphiti
        self._state = _load_skill_state()

    # ──────────────────────────────────────────────────────────────────────────
    # Public API
    # ──────────────────────────────────────────────────────────────────────────

    def execute_skill(self, skill_id: str, params: dict) -> dict:
        """
        Load skill, execute with Gemini vision, store to Graphiti.
        Synchronous entry point — runs the async executor internally.
        """
        if not PLAYWRIGHT_AVAILABLE:
            return {
                "success": False,
                "error": "Playwright not installed. Run: pip install playwright && playwright install chromium",
                "skill_id": skill_id,
            }

        try:
            skill = _load_skill(skill_id)
        except FileNotFoundError as e:
            return {"success": False, "error": str(e), "skill_id": skill_id}

        # Check Ebbinghaus decay
        if self._needs_reverification(skill_id, skill):
            log.warning(
                "Skill '%s' not used in %d days — flagged for re-verification",
                skill_id, skill.get("ebbinghaus_verify_days", 30),
            )

        # First try to recall from Graphiti
        recalled = self.recall_skill_from_graphiti(skill.get("platform", ""), skill_id)
        if recalled.get("found"):
            log.info("Using Graphiti-recalled workflow for '%s'", skill_id)
            # Merge recalled steps into skill if improved
            if recalled.get("refined_steps"):
                skill["steps"] = recalled["refined_steps"]

        log.info("Executing skill '%s' with params: %s", skill_id, list(params.keys()))
        result = asyncio.run(self._async_execute_skill(skill, params))

        # Update run statistics
        self._update_skill_stats(skill_id, result["success"])

        # Store to Graphiti on success
        if result["success"]:
            self._store_to_graphiti(skill, params, result)

        # Update skill map on failure
        if not result["success"] and result.get("failure_info"):
            self.update_skill_from_failure(skill_id, result["failure_info"])

        return result

    def learn_new_skill(self, platform: str, goal: str) -> dict:
        """
        Have Gemini explore a platform UI and map a new workflow skill.
        Generates a new skill JSON file from an exploratory session.
        """
        if not PLAYWRIGHT_AVAILABLE or not GEMINI_AVAILABLE:
            return {
                "success": False,
                "error": "Both Playwright and Gemini are required for skill learning",
            }

        log.info("Learning new skill for platform='%s', goal='%s'", platform, goal)
        result = asyncio.run(self._async_learn_skill(platform, goal))
        return result

    def update_skill_from_failure(self, skill_id: str, failure_info: dict) -> None:
        """
        Surprise gate: update workflow map when a step fails.
        Increments failure count, logs failure context, optionally updates steps.
        """
        try:
            skill = _load_skill(skill_id)
        except FileNotFoundError:
            log.warning("Cannot update non-existent skill '%s'", skill_id)
            return

        # Update skill metadata
        failed_step = failure_info.get("step", 0)
        failure_reason = failure_info.get("reason", "unknown")
        url_at_failure = failure_info.get("url", "")

        if "failure_log" not in skill:
            skill["failure_log"] = []

        skill["failure_log"].append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "step": failed_step,
            "reason": failure_reason,
            "url": url_at_failure,
        })

        # Keep last 10 failure records
        skill["failure_log"] = skill["failure_log"][-10:]

        # Append a failure note to the relevant step
        for step in skill.get("steps", []):
            if step.get("step") == failed_step:
                if "failure_notes" not in step:
                    step["failure_notes"] = []
                step["failure_notes"].append(
                    f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d')}] {failure_reason}"
                )

        _save_skill(skill)
        log.info("Skill '%s' failure map updated for step %d", skill_id, failed_step)

    def recall_skill_from_graphiti(self, platform: str, goal: str) -> dict:
        """
        Search Graphiti for previously learned browser workflows.
        Returns recalled episode data if found.
        """
        query = f"browser skill {platform} {goal} automation steps workflow"
        episodes = self.graphiti.search_episodes(query, limit=3)

        if not episodes:
            return {"found": False}

        log.info("Recalled %d Graphiti episodes for '%s/%s'", len(episodes), platform, goal)
        return {
            "found": True,
            "episodes": episodes,
            "refined_steps": None,  # Future: parse episodes back into step format
        }

    # ──────────────────────────────────────────────────────────────────────────
    # Listing / introspection
    # ──────────────────────────────────────────────────────────────────────────

    def list_skills(self, platform: Optional[str] = None) -> list[dict]:
        """Return all available skills with their metadata summaries."""
        skills = _list_skills(platform)
        state = _load_skill_state()
        summaries = []
        for s in skills:
            sid = s["skill_id"]
            run_data = state.get(sid, {})
            needs_reverify = self._needs_reverification(sid, s)
            summaries.append({
                "skill_id": sid,
                "platform": s.get("platform"),
                "description": s.get("description", "")[:100],
                "requires_auth": s.get("requires_auth", False),
                "requires_human_oversight": s.get("requires_human_oversight", False),
                "step_count": len(s.get("steps", [])),
                "success_rate": run_data.get("success_rate", s.get("success_rate", 0.0)),
                "execution_count": run_data.get("execution_count", 0),
                "last_run": run_data.get("last_run", "never"),
                "needs_reverification": needs_reverify,
                "last_verified": s.get("last_verified", "unknown"),
                "tags": s.get("tags", []),
            })
        return summaries

    def get_skill(self, skill_id: str) -> dict:
        """Return full skill definition."""
        return _load_skill(skill_id)

    # ──────────────────────────────────────────────────────────────────────────
    # Internal helpers
    # ──────────────────────────────────────────────────────────────────────────

    def _needs_reverification(self, skill_id: str, skill: dict) -> bool:
        """Check Ebbinghaus decay — has skill been dormant too long?"""
        state = self._state.get(skill_id, {})
        last_run = state.get("last_run")
        if not last_run or last_run == "never":
            return False  # Never run = new, not decayed
        try:
            last_dt = datetime.fromisoformat(last_run)
            decay_days = skill.get("ebbinghaus_verify_days", 30)
            return (datetime.now(timezone.utc) - last_dt).days > decay_days
        except Exception:
            return False

    def _update_skill_stats(self, skill_id: str, success: bool) -> None:
        """Update mutable execution statistics in state DB."""
        state = _load_skill_state()
        if skill_id not in state:
            state[skill_id] = {
                "execution_count": 0,
                "success_count": 0,
                "success_rate": 0.0,
                "last_run": None,
            }
        rec = state[skill_id]
        rec["execution_count"] += 1
        if success:
            rec["success_count"] = rec.get("success_count", 0) + 1
        total = rec["execution_count"]
        rec["success_rate"] = round(rec.get("success_count", 0) / total, 3)
        rec["last_run"] = datetime.now(timezone.utc).isoformat()
        self._state = state
        _save_skill_state(state)

    def _store_to_graphiti(self, skill: dict, params: dict, result: dict) -> None:
        """Store a successful execution as a Graphiti episode."""
        skill_id = skill["skill_id"]
        platform = skill.get("platform", "unknown")
        screenshot_hashes = result.get("screenshot_hashes", [])

        episode_body = (
            f"Browser skill '{skill_id}' executed successfully on platform '{platform}'.\n"
            f"Goal: {skill.get('description', '')}\n"
            f"Steps completed: {result.get('steps_completed', 0)} of {len(skill.get('steps', []))}\n"
            f"Params used: {json.dumps({k: '***' if 'password' in k.lower() else v for k, v in params.items()})}\n"
            f"Screenshot hashes: {', '.join(screenshot_hashes)}\n"
            f"Output fields: {json.dumps(result.get('outputs', {}))}\n"
            f"Timestamp: {datetime.now(timezone.utc).isoformat()}\n"
            f"Action history: {json.dumps(result.get('action_history', []))}"
        )

        episode_result = self.graphiti.store_episode(
            name=f"browser_skill_{skill_id}_{datetime.now().strftime('%Y%m%d')}",
            body=episode_body,
            source="browser_skill_engine",
        )

        if episode_result.get("stored"):
            # Also update the skill's graphiti_episodes list
            try:
                skill_data = _load_skill(skill_id)
                if "graphiti_episodes" not in skill_data:
                    skill_data["graphiti_episodes"] = []
                skill_data["graphiti_episodes"].append({
                    "episode_uuid": episode_result.get("episode_uuid"),
                    "timestamp": datetime.now(timezone.utc).isoformat(),
                })
                skill_data["graphiti_episodes"] = skill_data["graphiti_episodes"][-20:]
                _save_skill(skill_data)
            except Exception as e:
                log.warning("Could not update graphiti_episodes in skill file: %s", e)

    # ──────────────────────────────────────────────────────────────────────────
    # Async execution core
    # ──────────────────────────────────────────────────────────────────────────

    async def _async_execute_skill(self, skill: dict, params: dict) -> dict:
        """Core async skill executor — runs all steps via Playwright + Gemini."""
        skill_id = skill["skill_id"]
        steps = skill.get("steps", [])
        env_vars = skill.get("auth_env_vars", [])
        start_time = time.time()

        action_history: list[str] = []
        screenshot_hashes: list[str] = []
        outputs: dict = {}
        steps_completed = 0
        failure_info = None

        async with async_playwright() as playwright:
            try:
                browser, context, page = await _setup_browser(playwright)
            except Exception as e:
                return {
                    "success": False,
                    "skill_id": skill_id,
                    "error": f"Browser launch failed: {e}",
                    "steps_completed": 0,
                }

            try:
                for step_def in steps:
                    step_num = step_def.get("step", steps_completed + 1)
                    action_type = step_def.get("action", "vision_guided")
                    log.info("Executing step %d (%s)...", step_num, action_type)

                    if action_type == "navigate":
                        target = _resolve_params(
                            step_def.get("target", ""), params, env_vars
                        )
                        try:
                            await page.goto(
                                target, wait_until="domcontentloaded",
                                timeout=PLAYWRIGHT_TIMEOUT
                            )
                            action_history.append(f"Step {step_num}: navigated to {target}")
                            steps_completed += 1
                        except Exception as e:
                            failure_info = {
                                "step": step_num,
                                "reason": f"Navigation failed: {e}",
                                "url": target,
                            }
                            break

                    elif action_type in ("vision_guided", "extract_text"):
                        instruction = step_def.get("instruction", "")
                        success, desc, action_history = await _run_vision_step(
                            page, instruction, env_vars, params,
                            action_history, step_num,
                        )
                        # Capture screenshot hash after each vision step
                        _, h = await _take_screenshot_b64(page)
                        screenshot_hashes.append(h)

                        if success:
                            steps_completed += 1
                            # Extract outputs from extract steps
                            if action_type == "extract_text" and "extracted:" in desc.lower():
                                extracted_val = desc.split("Extracted:", 1)[-1].strip()
                                output_fields = skill.get("output_fields", [])
                                if output_fields:
                                    outputs[output_fields[0]] = extracted_val
                        else:
                            failure_info = {
                                "step": step_num,
                                "reason": desc,
                                "url": page.url,
                            }
                            log.warning(
                                "Skill '%s' failed at step %d: %s",
                                skill_id, step_num, desc
                            )
                            break

                    else:
                        log.warning(
                            "Unknown step action_type '%s' in skill '%s' step %d",
                            action_type, skill_id, step_num
                        )
                        steps_completed += 1  # Skip unknown steps gracefully

                # Final screenshot
                _, final_hash = await _take_screenshot_b64(page)
                screenshot_hashes.append(final_hash)

            except Exception as e:
                log.error("Unexpected error in skill '%s': %s", skill_id, e, exc_info=True)
                failure_info = {"step": steps_completed, "reason": str(e), "url": ""}
            finally:
                await context.close()
                await browser.close()

        success = failure_info is None and steps_completed == len(steps)
        elapsed = round(time.time() - start_time, 2)

        return {
            "success": success,
            "skill_id": skill_id,
            "steps_completed": steps_completed,
            "total_steps": len(steps),
            "outputs": outputs,
            "action_history": action_history,
            "screenshot_hashes": screenshot_hashes,
            "elapsed_seconds": elapsed,
            "failure_info": failure_info,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }

    async def _async_learn_skill(self, platform: str, goal: str) -> dict:
        """
        Exploratory skill learning session.
        Gemini explores the platform UI and maps a workflow.
        """
        skill_id = f"{platform.lower().replace(' ', '_')}_{goal.lower().replace(' ', '_')[:30].replace(' ', '_')}"
        log.info("Starting skill learning session: %s / %s", platform, goal)

        action_history: list[str] = []
        discovered_steps: list[dict] = []

        async with async_playwright() as playwright:
            try:
                browser, context, page = await _setup_browser(playwright)
            except Exception as e:
                return {"success": False, "error": f"Browser launch failed: {e}"}

            try:
                # Start from a known platform URL if we have it
                platform_urls = {
                    "gohighlevel": "https://app.gohighlevel.com/",
                    "telnyx": "https://portal.telnyx.com/",
                    "elestio": "https://elest.io/dashboard",
                    "godaddy": "https://sso.godaddy.com/",
                    "stripe": "https://dashboard.stripe.com/",
                    "netlify": "https://app.netlify.com/",
                    "n8n": os.environ.get("N8N_BASE_URL", "https://n8n-genesis-u50607.vm.elestio.app"),
                }
                start_url = platform_urls.get(platform.lower(), f"https://{platform}.com/")

                await page.goto(start_url, wait_until="domcontentloaded", timeout=PLAYWRIGHT_TIMEOUT)
                discovered_steps.append({"step": 1, "action": "navigate", "target": start_url})

                explore_prompt = (
                    f"You are mapping a browser workflow to accomplish: '{goal}' on {platform}. "
                    f"Explore the UI and identify the steps needed. "
                    f"Take note of navigation paths, button locations, and form fields required. "
                    f"This is step 2 — what do you see and what should be the first action?"
                )

                for step_num in range(2, 15):  # Max 13 exploration steps
                    screenshot_b64, _ = await _take_screenshot_b64(page)
                    action = await _gemini_decide_action(
                        screenshot_b64, explore_prompt, action_history, page.url, step_num
                    )

                    if action.get("action_type") == "done":
                        log.info("Skill mapping complete at step %d", step_num)
                        break

                    if action.get("action_type") == "fail":
                        log.warning("Learning failed at step %d: %s", step_num, action.get("reasoning"))
                        break

                    # Record as a skill step
                    discovered_steps.append({
                        "step": step_num,
                        "action": "vision_guided",
                        "instruction": action.get("reasoning", f"Step {step_num} of {goal}"),
                        "gemini_action": action,
                    })

                    # Execute the action to continue exploration
                    success, desc = await _execute_step_action(page, action, step_num)
                    action_history.append(f"Step {step_num}: {action.get('action_type')} — {desc}")

                    if not success:
                        log.warning("Exploration step %d failed: %s", step_num, desc)

                    await asyncio.sleep(0.8)

                    # Update prompt with what we learned
                    explore_prompt = (
                        f"Goal: '{goal}' on {platform}. "
                        f"Previous actions: {json.dumps(action_history[-5:])}. "
                        f"What is the next step to complete this goal?"
                    )

            finally:
                await context.close()
                await browser.close()

        # Build the new skill JSON
        new_skill = {
            "skill_id": skill_id,
            "platform": platform.lower(),
            "description": f"Auto-learned: {goal}",
            "url": platform_urls.get(platform.lower(), f"https://{platform}.com/"),
            "requires_auth": True,
            "auth_env_vars": [],
            "params": {},
            "steps": discovered_steps,
            "memory_key": f"{skill_id}_workflow",
            "last_verified": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
            "success_rate": 0.0,
            "execution_count": 0,
            "avg_steps_taken": 0,
            "graphiti_episodes": [],
            "auto_learned": True,
            "learning_timestamp": datetime.now(timezone.utc).isoformat(),
            "ebbinghaus_verify_days": 14,
            "tags": [platform.lower(), "auto-learned"],
        }

        skill_file = SKILLS_DIR / f"{skill_id}.json"
        with open(skill_file, "w", encoding="utf-8") as f:
            json.dump(new_skill, f, indent=2)

        log.info("New skill '%s' saved to %s", skill_id, skill_file)

        # Store learning session to Graphiti
        self.graphiti.store_episode(
            name=f"skill_learning_{skill_id}",
            body=(
                f"Learned new browser skill '{skill_id}' for goal: {goal}\n"
                f"Platform: {platform}\n"
                f"Steps discovered: {len(discovered_steps)}\n"
                f"Action history: {json.dumps(action_history)}"
            ),
        )

        return {
            "success": True,
            "skill_id": skill_id,
            "skill_file": str(skill_file),
            "steps_discovered": len(discovered_steps),
            "skill": new_skill,
        }


# =============================================================================
# MCP tool registration helpers (called by genesis-core MCP server)
# =============================================================================

_engine_singleton: Optional[BrowserSkillEngine] = None


def get_engine() -> BrowserSkillEngine:
    """Return singleton BrowserSkillEngine instance."""
    global _engine_singleton
    if _engine_singleton is None:
        _engine_singleton = BrowserSkillEngine()
    return _engine_singleton


def mcp_execute_browser_skill(skill_id: str, params: dict) -> dict:
    """MCP tool: Execute a browser skill by ID with given parameters."""
    return get_engine().execute_skill(skill_id, params)


def mcp_list_available_skills(platform: Optional[str] = None) -> list[dict]:
    """MCP tool: List all available browser skills, optionally filtered by platform."""
    return get_engine().list_skills(platform)


def mcp_learn_browser_skill(platform: str, goal: str) -> dict:
    """MCP tool: Explore a platform UI and learn a new browser workflow skill."""
    return get_engine().learn_new_skill(platform, goal)


# =============================================================================
# CLI interface
# =============================================================================

def _print_skills_table(skills: list[dict], verbose: bool = False) -> None:
    """Print skills in a formatted table."""
    platforms = {}
    for s in skills:
        p = s.get("platform", "unknown")
        if p not in platforms:
            platforms[p] = []
        platforms[p].append(s)

    total = len(skills)
    print(f"\nGenesis Browser Skills Registry — {total} skill(s)\n")
    print(f"{'SKILL ID':<45} {'PLATFORM':<15} {'STEPS':>5} {'RUNS':>5} {'SUCCESS%':>9} {'VERIFY?':<8}")
    print("-" * 95)

    for platform, pskills in sorted(platforms.items()):
        for s in pskills:
            sr = s.get("success_rate", 0.0)
            sr_str = f"{sr*100:.0f}%" if s.get("execution_count", 0) > 0 else "n/a"
            reverify = "YES" if s.get("needs_reverification") else ""
            oversight = " [HUMAN]" if s.get("requires_human_oversight") else ""
            print(
                f"  {s['skill_id']:<43} {platform:<15} {s.get('step_count', 0):>5} "
                f"{s.get('execution_count', 0):>5} {sr_str:>9} {reverify:<8}{oversight}"
            )
            if verbose:
                print(f"    {s.get('description', '')[:90]}")
                print(f"    Tags: {', '.join(s.get('tags', []))}")
                print()

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Genesis Browser Skill Engine",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 core/browser_skill_engine.py --list
  python3 core/browser_skill_engine.py --list --platform gohighlevel
  python3 core/browser_skill_engine.py --list --verbose
  python3 core/browser_skill_engine.py --execute ghl_create_subaccount_api_integration \\
      --params '{"subaccount_name": "Test Co", "integration_name": "Genesis"}'
  python3 core/browser_skill_engine.py --learn gohighlevel "set up email campaign"
  python3 core/browser_skill_engine.py --show ghl_deploy_snapshot
        """,
    )

    parser.add_argument("--list", "-l", action="store_true", help="List all available skills")
    parser.add_argument("--platform", "-p", help="Filter by platform (e.g. gohighlevel, telnyx)")
    parser.add_argument("--verbose", "-v", action="store_true", help="Show descriptions and tags")
    parser.add_argument("--execute", "-e", metavar="SKILL_ID", help="Execute a skill")
    parser.add_argument("--params", metavar="JSON", default="{}", help="JSON params for --execute")
    parser.add_argument("--learn", "-L", nargs=2, metavar=("PLATFORM", "GOAL"), help="Learn new skill")
    parser.add_argument("--show", "-s", metavar="SKILL_ID", help="Show full skill definition")
    parser.add_argument("--check-graphiti", action="store_true", help="Test Graphiti connection")
    parser.add_argument("--dry-run", action="store_true", help="Load skill and show steps without executing")

    args = parser.parse_args()
    engine = BrowserSkillEngine()

    if args.list or (not any([args.execute, args.learn, args.show, args.check_graphiti])):
        skills = engine.list_skills(args.platform)
        _print_skills_table(skills, verbose=args.verbose)
        if not PLAYWRIGHT_AVAILABLE:
            print("WARNING: Playwright not installed — execution disabled.")
            print("         Run: pip install playwright && playwright install chromium\n")
        if not GEMINI_AVAILABLE:
            print("WARNING: Gemini not configured — vision guidance disabled.")
            print("         Set GEMINI_API_KEY in config/secrets.env\n")

    elif args.show:
        try:
            skill = engine.get_skill(args.show)
            print(json.dumps(skill, indent=2))
        except FileNotFoundError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)

    elif args.execute:
        if args.dry_run:
            try:
                skill = engine.get_skill(args.execute)
                print(f"\nDRY RUN: Skill '{args.execute}'")
                print(f"Platform: {skill.get('platform')}")
                print(f"Description: {skill.get('description')}")
                print(f"\nSteps ({len(skill.get('steps', []))}):")
                for s in skill.get("steps", []):
                    print(f"  {s['step']:2}. [{s['action']}] {s.get('instruction', s.get('target', ''))}")
            except FileNotFoundError as e:
                print(f"Error: {e}", file=sys.stderr)
                sys.exit(1)
        else:
            try:
                params = json.loads(args.params)
            except json.JSONDecodeError as e:
                print(f"Invalid JSON params: {e}", file=sys.stderr)
                sys.exit(1)
            print(f"\nExecuting skill '{args.execute}'...")
            result = engine.execute_skill(args.execute, params)
            print(json.dumps(result, indent=2))
            sys.exit(0 if result.get("success") else 1)

    elif args.learn:
        platform, goal = args.learn
        print(f"\nLearning new skill for: {platform} / {goal}")
        result = engine.learn_new_skill(platform, goal)
        print(json.dumps(result, indent=2))

    elif args.check_graphiti:
        print(f"\nChecking Graphiti connection at {GRAPHITI_MCP_URL}...")
        available = engine.graphiti._check_available()
        print(f"Graphiti available: {available}")
        if available:
            episodes = engine.graphiti.search_episodes("browser skill test", limit=2)
            print(f"Test search returned {len(episodes)} results")


if __name__ == "__main__":
    main()