AIPass/src/aipass/hooks/apps/modules/engine.py

# =================== AIPass ====================
# Name: engine.py
# Version: 1.1.0
# Description: Hook engine — unified dispatcher for all hook events
# Branch: hooks
# Layer: apps/modules
# Created: 2026-05-18
# Modified: 2026-05-19
# =============================================

"""Hook engine — dispatches hook events to handlers, logs via prax + JSONL."""

import importlib
import json
import os
import subprocess
import tempfile
import time
from pathlib import Path

from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import err_console
from aipass.hooks.apps.handlers.config.diagnostics import log_entry as _log, tail_log

CONSOLE = err_console
BRANCH_ROOT = Path(__file__).resolve().parent.parent.parent

HELP_COMMANDS = [
    ("log", "Tail recent hook activity (last 20 entries)"),
]


def _run_hook(hook_cmd: str, stdin_data: str, timeout_s: int = 30) -> dict:
    """Run a single hook subprocess, capture output and timing."""
    env = os.environ.copy()
    start = time.monotonic()
    try:
        result = subprocess.run(
            hook_cmd,
            shell=True,
            input=stdin_data,
            capture_output=True,
            text=True,
            timeout=timeout_s,
            env=env,
        )
        elapsed_ms = (time.monotonic() - start) * 1000
        return {
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "elapsed_ms": round(elapsed_ms, 1),
        }
    except subprocess.TimeoutExpired:
        elapsed_ms = (time.monotonic() - start) * 1000
        logger.error("[HOOKS] timeout after %ds: %s", timeout_s, hook_cmd)
        return {"exit_code": -1, "stdout": "", "stderr": "TIMEOUT", "elapsed_ms": round(elapsed_ms, 1)}
    except OSError as exc:
        elapsed_ms = (time.monotonic() - start) * 1000
        logger.error("[HOOKS] exec error: %s: %s", hook_cmd, exc)
        return {"exit_code": -1, "stdout": "", "stderr": str(exc), "elapsed_ms": round(elapsed_ms, 1)}


def _run_handler(handler_path: str, hook_data: dict) -> dict:
    """Call a handler function directly (no subprocess). Module imports handler."""
    start = time.monotonic()
    try:
        module_path, func_name = handler_path.rsplit(".", 1)
        if not module_path.startswith("aipass."):
            elapsed_ms = (time.monotonic() - start) * 1000
            logger.warning(
                "[HOOKS] handler path refused (not in aipass.* namespace): %s",
                handler_path,
            )
            return {
                "exit_code": -1,
                "stdout": "",
                "stderr": f"handler namespace refused: {handler_path}",
                "elapsed_ms": round(elapsed_ms, 1),
            }
        module = importlib.import_module(module_path)
        handler_func = getattr(module, func_name)
        result = handler_func(hook_data)
        elapsed_ms = (time.monotonic() - start) * 1000
        return {
            "exit_code": result.get("exit_code", 0),
            "stdout": result.get("stdout", ""),
            "sound": result.get("sound", ""),
            "stderr": "",
            "elapsed_ms": round(elapsed_ms, 1),
        }
    except Exception as exc:
        elapsed_ms = (time.monotonic() - start) * 1000
        logger.error("[HOOKS] handler error %s: %s", handler_path, exc)
        return {"exit_code": -1, "stdout": "", "stderr": str(exc), "elapsed_ms": round(elapsed_ms, 1)}


def _matches(matcher: str, value: str) -> bool:
    """Check if a hook's matcher string matches the given value. Empty matcher = always match."""
    if not matcher:
        return True
    return value in matcher.split("|")


_BUDGET_KEYS = ("max_per_session", "min_spacing_turns", "cooldown_seconds")


def _budget_state_path(session_id: str = "") -> Path | None:
    if not session_id:
        session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "")
    if not session_id:
        return None
    return Path(tempfile.gettempdir()) / f"aipass-handler-budget-{session_id}.json"


def _load_budget_state(session_id: str = "") -> dict:
    path = _budget_state_path(session_id)
    if path is None or not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError) as exc:
        logger.info("[HOOKS] budget: state read failed: %s", exc)
        return {}


def _save_budget_state(state: dict, session_id: str = "") -> None:
    path = _budget_state_path(session_id)
    if path is None:
        return
    try:
        path.write_text(json.dumps(state), encoding="utf-8")
    except OSError as exc:
        logger.info("[HOOKS] budget: state write failed: %s", exc)


def _check_budget(hook_name: str, budget_cfg: dict, budget_state: dict) -> tuple[bool, str]:
    """Check if handler is within its per-session budget."""
    hs = budget_state.get(hook_name, {})
    fire_count = hs.get("fire_count", 0)

    max_fires = budget_cfg.get("max_per_session")
    if max_fires is not None and fire_count >= max_fires:
        return False, f"budget exhausted ({fire_count}/{max_fires})"

    if fire_count > 0:
        min_spacing = budget_cfg.get("min_spacing_turns")
        if min_spacing is not None:
            turns_since = hs.get("turns_since_fire", 0)
            if turns_since < min_spacing:
                return False, f"spacing ({turns_since}/{min_spacing})"

        cooldown = budget_cfg.get("cooldown_seconds")
        if cooldown is not None:
            elapsed = time.time() - hs.get("last_fire_time", 0.0)
            if elapsed < cooldown:
                return False, f"cooldown ({int(cooldown - elapsed)}s)"

    return True, "ok"


def dispatch(event_type: str, stdin_data: str, config: dict) -> tuple[str, int]:
    """Core dispatch — run hooks for event, return (merged_stdout, exit_code)."""
    if not config.get("hooks_enabled", True):
        logger.info("[HOOKS] all hooks disabled")
        _log({"ts": time.time(), "event": event_type, "action": "all_hooks_disabled"})
        return "", 0

    event_hooks = config.get(event_type, {})
    if not event_hooks:
        _log({"ts": time.time(), "event": event_type, "action": "no_hooks_configured"})
        return "", 0

    match_value = ""
    parsed = {}
    try:
        parsed = json.loads(stdin_data) if stdin_data.strip() else {}
        match_value = parsed.get("tool_name", "") or parsed.get("compact_type", "") or parsed.get("type", "")
    except json.JSONDecodeError as exc:
        logger.warning("[HOOKS] stdin parse error: %s", exc)

    outputs = []
    total_start = time.monotonic()
    budget_state = None
    budget_dirty = False
    payload_session_id = parsed.get("session_id", "")

    for hook_name, hook_def in event_hooks.items():
        if not hook_def.get("enabled", True):
            logger.info("[HOOKS] %s.%s skipped (disabled)", event_type, hook_name)
            _log({"ts": time.time(), "event": event_type, "hook": hook_name, "action": "skipped_disabled"})
            continue

        handler = hook_def.get("handler", "")
        command = hook_def.get("command", "")
        matcher = hook_def.get("matcher", "")
        if not handler and not command:
            continue

        if matcher and not _matches(matcher, match_value):
            _log(
                {
                    "ts": time.time(),
                    "event": event_type,
                    "hook": hook_name,
                    "action": "skipped_no_match",
                    "matcher": matcher,
                    "value": match_value,
                }
            )
            continue

        if command and not handler and config.get("_source") == "project":
            logger.warning(
                "[HOOKS] %s.%s REFUSED: command-type not allowed in per-project config",
                event_type,
                hook_name,
            )
            _log(
                {
                    "ts": time.time(),
                    "event": event_type,
                    "hook": hook_name,
                    "action": "refused_command_type",
                }
            )
            continue

        budget_cfg = {k: hook_def[k] for k in _BUDGET_KEYS if k in hook_def}
        if budget_cfg:
            if budget_state is None:
                budget_state = _load_budget_state(payload_session_id)
            hs = budget_state.setdefault(hook_name, {})
            hs["turns_since_fire"] = hs.get("turns_since_fire", 0) + 1
            budget_dirty = True
            allowed, reason = _check_budget(hook_name, budget_cfg, budget_state)
            if not allowed:
                logger.info("[HOOKS] %s.%s budget: %s", event_type, hook_name, reason)
                _log(
                    {
                        "ts": time.time(),
                        "event": event_type,
                        "hook": hook_name,
                        "action": "budget_suppressed",
                        "reason": reason,
                    }
                )
                continue

        if handler:
            result = _run_handler(handler, parsed)
        else:
            hook_timeout = hook_def.get("timeout", 30)
            result = _run_hook(command, stdin_data, timeout_s=hook_timeout)

        logger.info(
            "[HOOKS] %s.%s agent=%s exit=%d out=%db %dms",
            event_type,
            hook_name,
            parsed.get("agent_type", "") or "main",
            result["exit_code"],
            len(result["stdout"]),
            result["elapsed_ms"],
        )
        _log(
            {
                "ts": time.time(),
                "event": event_type,
                "hook": hook_name,
                "agent_type": parsed.get("agent_type", ""),
                "agent_id": parsed.get("agent_id", ""),
                "exit_code": result["exit_code"],
                "elapsed_ms": result["elapsed_ms"],
                "stdout_len": len(result["stdout"]),
                "stderr_preview": result["stderr"][:200] if result["stderr"] else "",
                "cwd": str(Path.cwd()),
            }
        )

        if result.get("sound"):
            try:
                from aipass.hooks.apps.sound import speak

                speak(result["sound"])
            except Exception as exc:
                logger.info("[HOOKS] sound playback failed for %s.%s: %s", event_type, hook_name, exc)

        # Exit code 2: crash vs intentional block
        if result["exit_code"] == 2:
            is_intentional_block = False
            try:
                decision = json.loads(result["stdout"]) if result["stdout"].strip() else {}
                is_intentional_block = decision.get("decision") == "block"
            except (json.JSONDecodeError, AttributeError):
                logger.info("[HOOKS] %s.%s exit=2 stdout not JSON, treating as crash", event_type, hook_name)

            if is_intentional_block:
                total_ms = (time.monotonic() - total_start) * 1000
                logger.warning("[HOOKS] %s BLOCKED by %s (%dms)", event_type, hook_name, total_ms)
                _log(
                    {
                        "ts": time.time(),
                        "event": event_type,
                        "action": "blocked",
                        "hook": hook_name,
                        "total_ms": round(total_ms, 1),
                    }
                )
                return result["stdout"], 2

            logger.error(
                "[HOOKS] %s.%s CRASHED exit=2: %s",
                event_type,
                hook_name,
                result["stderr"][:200],
            )
            _log(
                {
                    "ts": time.time(),
                    "event": event_type,
                    "hook": hook_name,
                    "action": "crashed",
                    "stderr": result["stderr"][:200],
                }
            )

        if result["stdout"]:
            outputs.append(result["stdout"])
            if budget_cfg and budget_state is not None:
                hs = budget_state.setdefault(hook_name, {})
                hs["fire_count"] = hs.get("fire_count", 0) + 1
                hs["last_fire_time"] = time.time()
                hs["turns_since_fire"] = 0
                budget_dirty = True

    if budget_dirty and budget_state is not None:
        _save_budget_state(budget_state, payload_session_id)

    total_ms = (time.monotonic() - total_start) * 1000
    logger.info("[HOOKS] %s complete: %d hooks %dms", event_type, len(outputs), total_ms)
    _log(
        {
            "ts": time.time(),
            "event": event_type,
            "action": "complete",
            "hooks_run": len(outputs),
            "total_ms": round(total_ms, 1),
        }
    )

    return "\n".join(outputs), 0


# =============================================================================
# MODULE INTERFACE (drone @hooks routing)
# =============================================================================


def print_introspection():
    """Print module structure — connected handlers."""
    CONSOLE.print("[bold cyan]engine[/bold cyan] Module")
    CONSOLE.print("  Connected Handlers:")
    handlers_root = BRANCH_ROOT / "apps" / "handlers"
    for category_dir in sorted(handlers_root.iterdir()):
        if not category_dir.is_dir() or category_dir.name.startswith("_"):
            continue
        handler_files = [f.name for f in sorted(category_dir.glob("*.py")) if not f.name.startswith("_")]
        if handler_files:
            CONSOLE.print(f"    handlers/{category_dir.name}/ — {', '.join(handler_files)}")


def handle_command(command: str, args: list) -> bool:
    """Route engine commands from drone @hooks."""
    if command in ("engine", ""):
        if not args:
            print_introspection()
            return True

    if command in ("--help", "-h", "help"):
        CONSOLE.print("[bold cyan]engine[/bold cyan] — Hook dispatch engine")
        CONSOLE.print()
        CONSOLE.print("  drone @hooks log       Tail recent hook activity")
        return True

    if command == "log":
        lines = tail_log(20)
        if not lines:
            CONSOLE.print("No engine log found")
        else:
            for line in lines:
                CONSOLE.print(line)
        return True

    return False