JARVIS/backend/app/routers/hooks.py

"""Hook API 路由 - Phase 7.5"""

from typing import Any

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from app.agents.tools.hooks import HookType
from app.agents.tools.hooks.builtins import (
    AuditLogHook,
    DangerousConfirmationHook,
    SecurityScanHook,
)
from app.agents.tools.hooks.config import (
    HookConfigEntry,
    get_hook_config_persistence,
)
from app.agents.tools.hooks.manager import get_hook_manager

router = APIRouter(prefix="/api/hooks", tags=["Hooks"])


class HookInfo(BaseModel):
    """Hook 信息"""

    name: str
    hook_type: str
    description: str
    builtin: bool


class HookConfigUpdate(BaseModel):
    """更新 Hook 配置"""

    entries: list[HookConfigEntry]


class HookConfigResponse(BaseModel):
    """Hook 配置响应"""

    entries: list[dict[str, Any]]
    count: int


class HookStatusResponse(BaseModel):
    """Hook 状态响应"""

    name: str
    enabled: bool
    hook_type: str
    registered: bool


# 内置 Hook 注册表
BUILTIN_HOOKS: dict[str, dict[str, str]] = {
    "audit_log": {
        "name": "audit_log",
        "hook_type": "pre_tool_use,post_tool_use,tool_error",
        "description": "审计日志 Hook - 记录所有工具调用",
        "class": "AuditLogHook",
    },
    "dangerous_confirmation": {
        "name": "dangerous_confirmation",
        "hook_type": "pre_tool_use",
        "description": "危险操作确认 Hook - 拦截危险工具调用",
        "class": "DangerousConfirmationHook",
    },
    "security_scan": {
        "name": "security_scan",
        "hook_type": "post_tool_use",
        "description": "安全扫描 Hook - 检测敏感信息泄露",
        "class": "SecurityScanHook",
    },
}


@router.get("/available", response_model=list[HookInfo])
async def list_available_hooks() -> list[HookInfo]:
    """列出所有可用的内置 Hook"""
    return [
        HookInfo(
            name=info["name"],
            hook_type=info["hook_type"],
            description=info["description"],
            builtin=True,
        )
        for info in BUILTIN_HOOKS.values()
    ]


@router.get("/config", response_model=HookConfigResponse)
async def get_hook_config() -> HookConfigResponse:
    """获取当前 Hook 配置"""
    persistence = get_hook_config_persistence()
    entries = persistence.load_config()
    return HookConfigResponse(
        entries=[vars(e) if isinstance(e, HookConfigEntry) else e for e in entries],
        count=len(entries),
    )


@router.post("/config", response_model=HookConfigResponse)
async def update_hook_config(
    entries: list[HookConfigEntry],
) -> HookConfigResponse:
    """更新 Hook 配置"""
    persistence = get_hook_config_persistence()
    success = persistence.save_config(entries)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to save hook config")

    # 应用配置到 HookManager
    manager = get_hook_manager()
    manager.clear()  # 清除旧配置
    persistence.apply_config()  # 应用新配置

    return HookConfigResponse(
        entries=[vars(e) if isinstance(e, HookConfigEntry) else e for e in entries],
        count=len(entries),
    )


@router.post("/apply-config", response_model=dict[str, Any])
async def apply_hook_config() -> dict[str, Any]:
    """应用配置文件到 HookManager"""
    persistence = get_hook_config_persistence()
    manager = get_hook_manager()
    manager.clear()
    count = persistence.apply_config()
    return {"applied": count, "message": f"Applied {count} hook configurations"}


@router.get("/status", response_model=list[HookStatusResponse])
async def get_hook_status() -> list[HookStatusResponse]:
    """获取所有已注册 Hook 的状态"""
    manager = get_hook_manager()
    all_hooks = manager.list_all()

    # 按名称索引已注册的 hooks
    registered: dict[str, dict[str, Any]] = {}
    for hook in all_hooks:
        registered[hook.name] = {
            "name": hook.name,
            "enabled": hook.enabled,
            "hook_type": hook.hook_type.value,
            "registered": True,
        }

    # 合并内置 Hook 信息
    result: list[HookStatusResponse] = []
    seen: set[str] = set()

    # 先添加已注册的
    for hook in all_hooks:
        result.append(
            HookStatusResponse(
                name=hook.name,
                enabled=hook.enabled,
                hook_type=hook.hook_type.value,
                registered=True,
            )
        )
        seen.add(hook.name)

    # 再添加内置但未注册的
    for name, info in BUILTIN_HOOKS.items():
        if name not in seen:
            result.append(
                HookStatusResponse(
                    name=name,
                    enabled=False,
                    hook_type=info["hook_type"],
                    registered=False,
                )
            )

    return result


@router.post("/{name}/enable", response_model=dict[str, str])
async def enable_hook(name: str) -> dict[str, str]:
    """启用指定 Hook"""
    manager = get_hook_manager()
    if manager.enable(name):
        return {"status": "enabled", "name": name}
    raise HTTPException(status_code=404, detail=f"Hook '{name}' not found")


@router.post("/{name}/disable", response_model=dict[str, str])
async def disable_hook(name: str) -> dict[str, str]:
    """禁用指定 Hook"""
    manager = get_hook_manager()
    if manager.disable(name):
        return {"status": "disabled", "name": name}
    raise HTTPException(status_code=404, detail=f"Hook '{name}' not found")


@router.post("/register-builtin", response_model=dict[str, str])
async def register_builtin_hook(
    name: str,
    hook_type: str = "pre_tool_use",
) -> dict[str, str]:
    """注册内置 Hook 到 HookManager"""
    from app.agents.tools.hooks.types import HookDefinition, HookTrigger

    manager = get_hook_manager()

    if name == "audit_log":
        hook_instance = AuditLogHook()
        handler = hook_instance.pre_tool_use
        hook_types = [HookType.PRE_TOOL_USE, HookType.POST_TOOL_USE, HookType.TOOL_ERROR]
    elif name == "dangerous_confirmation":
        hook_instance = DangerousConfirmationHook()
        handler = hook_instance.pre_tool_use
        hook_types = [HookType.PRE_TOOL_USE]
    elif name == "security_scan":
        hook_instance = SecurityScanHook()
        handler = hook_instance.post_tool_use
        hook_types = [HookType.POST_TOOL_USE]
    else:
        raise HTTPException(status_code=404, detail=f"Unknown builtin hook: {name}")

    registered = []
    for ht in hook_types:
        hook_def = HookDefinition(
            name=f"{name}_{ht.value}",
            hook_type=ht,
            trigger=HookTrigger(),
            handler=handler,
            priority=0,
            enabled=True,
            description=f"Built-in {name} hook",
        )
        manager.register(hook_def)
        registered.append(ht.value)

    return {
        "status": "registered",
        "name": name,
        "hook_types": ", ".join(registered),
    }