from dataclasses import dataclass
from typing import Any, Callable
import jsonschema, hashlib, time, logging


@dataclass
class ToolCall:
    name: str
    params: dict[str, Any]
    session_id: str


def _request_human_approval(call: ToolCall) -> str:
    """人工审批占位，实际系统中对接审批工作流。"""
    return f"[PENDING] 工具 '{call.name}' 等待人工审批——会话 {call.session_id}"


class CapabilityGate:
    """策略逻辑在初始化时从外部配置加载，LLM 运行期间不可修改。"""

    def __init__(self, policy: dict):
                self.policy = policy
        self.audit_log = []

    def execute(self, call: ToolCall) -> Any:
                if call.name not in self.policy["allowed_tools"]:
            self._audit(call, "BLOCKED_UNKNOWN_TOOL")
            raise PermissionError(f"工具 '{call.name}' 不在能力允许列表中")

        tool_policy = self.policy["allowed_tools"][call.name]

                try:
            jsonschema.validate(call.params, tool_policy["param_schema"])
        except jsonschema.ValidationError as e:
            self._audit(call, "BLOCKED_SCHEMA_VIOLATION", str(e))
            raise

                if tool_policy.get("requires_human_approval"):
            self._audit(call, "PENDING_HUMAN_REVIEW")
            return self._request_human_approval(call)

                self._audit(call, "EXECUTED")
        handler: Callable = tool_policy["handler"]
        return handler(**call.params)

    def _audit(self, call: ToolCall, outcome: str, detail: str = ""):
        entry = {
            "ts": time.time(),
            "session": call.session_id,
            "tool": call.name,
            "params_hash": hashlib.sha256(
                str(call.params).encode()
            ).hexdigest()[:16],
            "outcome": outcome,
        }
        self.audit_log.append(entry)
        logging.info(
            "[GATE] %s | %s | session=%s", call.name, outcome, call.session_id
        )