code/hooks/__init__.py

"""Hooks system — pre/post tool execution rules.

Inspired by Claude Code's hookify plugin. Rules are markdown files
with YAML frontmatter that define:
- event: bash | file | prompt | stop | all
- pattern: regex to match
- action: warn | block
- message: shown to the user/agent when triggered

Rules are discovered from:
- code/hooks/builtins/  (built-in rules)
- workspace/.sonicoder/hooks/  (user rules)
"""

from __future__ import annotations

import logging
import os
import re
from typing import Any

from code.skills import _parse_frontmatter

logger = logging.getLogger(__name__)

_BUILTIN_HOOKS_DIR = os.path.join(os.path.dirname(__file__), "builtins")
_USER_HOOKS_DIRNAME = ".sonicoder/hooks"


def _hook_dirs() -> list[str]:
    dirs = [_BUILTIN_HOOKS_DIR]
    try:
        from code.tools.fs import get_workspace_root
        user_dir = os.path.join(get_workspace_root(), _USER_HOOKS_DIRNAME)
        if os.path.isdir(user_dir):
            dirs.append(user_dir)
    except Exception:
        pass
    return dirs


def _load_hook(filepath: str) -> dict[str, Any] | None:
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception:
        return None

    meta, body = _parse_frontmatter(content)

    name = meta.get("name", os.path.splitext(os.path.basename(filepath))[0])
    enabled = meta.get("enabled", "true").lower() == "true"
    event = meta.get("event", "all").lower()
    action = meta.get("action", "warn").lower()
    pattern = meta.get("pattern", "")
    conditions_raw = meta.get("conditions", "")

    # Parse conditions (simplified — actual hookify uses YAML lists)
    conditions: list[dict[str, str]] = []
    if conditions_raw:
        # Very rough parse: each "- field: x\n  operator: y\n  pattern: z"
        for block in re.split(r"(?=\n-\s+field:)", "\n" + conditions_raw):
            field_m = re.search(r"field:\s*(\S+)", block)
            op_m = re.search(r"operator:\s*(\S+)", block)
            pat_m = re.search(r"pattern:\s*(.+?)(?=\n\s*$|\n\s*-|\Z)", block, re.DOTALL)
            if field_m and op_m and pat_m:
                conditions.append({
                    "field": field_m.group(1),
                    "operator": op_m.group(1),
                    "pattern": pat_m.group(1).strip(),
                })

    return {
        "name": name,
        "enabled": enabled,
        "event": event,
        "action": action,
        "pattern": pattern,
        "conditions": conditions,
        "message": body.strip(),
        "path": filepath,
    }


def list_hooks() -> list[dict[str, Any]]:
    """List all hooks (metadata only)."""
    hooks: list[dict[str, Any]] = []
    seen: set[str] = set()

    for hooks_dir in _hook_dirs():
        if not os.path.isdir(hooks_dir):
            continue
        for entry in sorted(os.listdir(hooks_dir)):
            if not entry.endswith(".md"):
                continue
            filepath = os.path.join(hooks_dir, entry)
            hook = _load_hook(filepath)
            if hook and hook["name"] not in seen:
                seen.add(hook["name"])
                hooks.append({
                    "name": hook["name"],
                    "enabled": hook["enabled"],
                    "event": hook["event"],
                    "action": hook["action"],
                    "pattern": hook["pattern"],
                })
    return hooks


def _match_condition(condition: dict[str, str], context: dict[str, Any]) -> bool:
    """Check if a single condition matches."""
    field = condition.get("field", "")
    operator = condition.get("operator", "regex_match")
    pattern = condition.get("pattern", "")

    value = str(context.get(field, ""))

    if operator == "regex_match":
        return bool(re.search(pattern, value))
    elif operator == "contains":
        return pattern in value
    elif operator == "equals":
        return value == pattern
    elif operator == "not_contains":
        return pattern not in value
    elif operator == "starts_with":
        return value.startswith(pattern)
    elif operator == "ends_with":
        return value.endswith(pattern)
    return False


def _match_hook(hook: dict[str, Any], event: str, context: dict[str, Any]) -> bool:
    """Check if a hook matches the given event and context."""
    if not hook["enabled"]:
        return False
    if hook["event"] != "all" and hook["event"] != event:
        return False

    # Simple pattern match (single pattern)
    if hook["pattern"]:
        # For bash event, match against command
        target = str(context.get("command", context.get("file_path", context.get("user_prompt", ""))))
        if not re.search(hook["pattern"], target):
            return False

    # Multi-condition match (all conditions must match)
    if hook["conditions"]:
        for cond in hook["conditions"]:
            if not _match_condition(cond, context):
                return False

    return True


def check_hook(event: str, context: dict[str, Any]) -> dict[str, Any]:
    """Check all hooks for an event.

    Args:
        event: One of 'bash', 'file', 'prompt', 'stop', 'all'
        context: Dict with relevant fields (command, file_path, new_text, user_prompt, etc.)

    Returns:
        dict with:
        - blocked: bool — whether the action should be blocked
        - warnings: list of warning messages to show
        - matched_hooks: list of hook names that matched
    """
    warnings: list[str] = []
    matched: list[str] = []
    blocked = False

    for hooks_dir in _hook_dirs():
        if not os.path.isdir(hooks_dir):
            continue
        for entry in sorted(os.listdir(hooks_dir)):
            if not entry.endswith(".md"):
                continue
            filepath = os.path.join(hooks_dir, entry)
            hook = _load_hook(filepath)
            if not hook:
                continue
            if _match_hook(hook, event, context):
                matched.append(hook["name"])
                if hook["action"] == "block":
                    blocked = True
                    warnings.append(f"🛑 BLOCKED by rule '{hook['name']}':\n\n{hook['message']}")
                else:
                    warnings.append(f"⚠️ Warning from rule '{hook['name']}':\n\n{hook['message']}")

    return {
        "blocked": blocked,
        "warnings": warnings,
        "matched_hooks": matched,
    }


def create_hook(
    name: str,
    event: str,
    pattern: str,
    action: str = "warn",
    message: str = "",
    enabled: bool = True,
) -> dict[str, Any]:
    """Create a new user hook (saved to workspace/.sonicoder/hooks/)."""
    try:
        from code.tools.fs import get_workspace_root
        hooks_dir = os.path.join(get_workspace_root(), _USER_HOOKS_DIRNAME)
        os.makedirs(hooks_dir, exist_ok=True)

        filepath = os.path.join(hooks_dir, f"{name}.local.md")
        content = f"""---
name: {name}
enabled: {str(enabled).lower()}
event: {event}
pattern: {pattern}
action: {action}
---

{message}
"""
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)

        return {"success": True, "name": name, "path": filepath}
    except Exception as exc:
        return {"success": False, "error": str(exc)}


def delete_hook(name: str) -> dict[str, Any]:
    """Delete a user hook by name."""
    try:
        from code.tools.fs import get_workspace_root
        hooks_dir = os.path.join(get_workspace_root(), _USER_HOOKS_DIRNAME)
        filepath = os.path.join(hooks_dir, f"{name}.local.md")
        if os.path.exists(filepath):
            os.remove(filepath)
            return {"success": True, "name": name}
        return {"success": False, "error": f"Hook not found: {name}"}
    except Exception as exc:
        return {"success": False, "error": str(exc)}