"""Project-scoped tool allowlists from ``project.yaml``."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any, cast
from butler.tools.project_tools_ops import dev_engine_extra_tools
from butler.tools.project_tools_ops import mcp_tool_allowed
from butler.tools.project_tools_ops import workflow_step_tool_allowlist
from butler.tools.registry import get_tool_definitions
from butler.tools.project_tools_ops import optimize_tool_definitions_safe
from butler.execution_context import get_current_session_key
from butler.project.manager import get_project_manager

if TYPE_CHECKING:
    from butler.project import Project

# Names in project.yaml may use legacy aliases from design docs.
_TOOL_NAME_ALIASES: dict[str, str] = {
    "edit_file": "patch",
    "search_code": "search_files",
    "run_shell": "terminal",
    "skill_list": "skills_list",
}

# Personal butler (no project): host read-only + recall; no workflow/delegate.
_PERSONAL_BUTLER_TOOLS = frozenset({
    "read_file",
    "search_files",
    "list_directory",
    "terminal",
    "butler_recall",
    "butler_remember",
    "search_transcript",
    "skills_list",
    "skill_view",
    "list_runtime_jobs",
})

# Butler orchestration tools always available when a project is selected.
_BUTLER_EXTRA_TOOLS = frozenset({
    "delegate_task",
    "list_workflows",
    "search_project_knowledge",
    "skills_list",
    "skill_view",
    "butler_remember",
    "butler_recall",
    "list_runtime_jobs",
    "run_runtime_job",
    "contact_add",
    "contact_find",
    "contact_update",
    "contact_delete",
    "contact_list",
    "memo_add",
    "memo_list",
    "memo_search",
    "memo_update",
    "memo_delete",
    "expense_add",
    "expense_summary",
    "expense_list",
    "expense_update",
    "expense_search",
    "expense_delete",
    "habit_create",
    "habit_checkin",
    "habit_stats",
    "habit_list",
    "habit_update",
    "habit_delete",
    "set_reminder",
    "list_reminders",
    "reminder_list_active",
    "cancel_reminder",
})

# Project Lead: read-only + orchestration (no write/shell even if listed in project.yaml).
_LEAD_READ_TOOLS = frozenset({
    "read_file",
    "list_directory",
    "search_files",
})
_PLAN_MODE_TOOLS = frozenset({
    "read_file",
    "list_directory",
    "search_files",
    "search_project_knowledge",
    "search_transcript",
    "skills_list",
    "skill_view",
    "butler_remember",
    "butler_recall",
})

_LEAD_EXTRA_TOOLS = frozenset({
    "delegate_task",
    "run_workflow",
    "list_workflows",
    "skills_list",
    "skill_view",
    "butler_remember",
    "butler_recall",
    "list_runtime_jobs",
    "run_runtime_job",
})

# Lead may opt in read-only network tools from project.yaml (no shell / no fetch by default).
_LEAD_NETWORK_TOOLS = frozenset({
    "web_search",
})

_DEV_EXTRA_TOOLS = frozenset({
    "dev_status",
    "dev_verify",
    "dev_review",
    "dev_rollback",
    "dev_search_symbols",
    "run_pytest",
})

# Lead (厂长) 对话引擎：只读 + 编排；改盘仍走 delegate。
# Butler (管家) 对话引擎：allowed=None，不受 project.yaml / wechat_minimal 裁剪。


def _mcp_allowlist_from_mapped(mapped: set[str]) -> set[str]:
    """Preserve project.yaml MCP opt-in (``mcp_*`` or explicit ``mcp_`` names)."""
    return {
        name
        for name in mapped
        if name == "mcp_*" or (isinstance(name, str) and name.startswith("mcp_"))
    }


def _lead_network_from_mapped(mapped: set[str]) -> set[str]:
    """Preserve project.yaml opt-in for read-only Lead network tools."""
    return {name for name in mapped if name in _LEAD_NETWORK_TOOLS}


def _lead_allowed_tools(mapped: set[str]) -> set[str]:
    """Lead read-only core + orchestration; honor MCP opt-in from project.yaml."""
    read_only = {n for n in mapped if n in _LEAD_READ_TOOLS}
    if not read_only:
        read_only = set(_LEAD_READ_TOOLS)
    return (
        read_only
        | set(_LEAD_EXTRA_TOOLS)
        | _mcp_allowlist_from_mapped(mapped)
        | _lead_network_from_mapped(mapped)
    )


def canonical_tool_name(name: str) -> str:
    """Map project.yaml tool name to registry tool name."""
    key = str(name or "").strip()
    if not key:
        return ""
    return _TOOL_NAME_ALIASES.get(key, key)


def allowed_tool_names_for_project(
    project: "Project | None",
    *,
    role: str = "butler",
) -> set[str] | None:
    """Return allowed registry tool names, or ``None`` if unrestricted."""
    norm_early = role.replace("_agent", "").strip().lower()
    if norm_early == "plan":
        return set(_PLAN_MODE_TOOLS)
    if norm_early == "lead":
        mapped: set[str] = set()
        if project is not None:
            raw = [canonical_tool_name(n) for n in (project.tools or [])]
            mapped = {n for n in raw if n}
            modes = getattr(project, "tool_modes", None) or {}
            if isinstance(modes, dict) and modes:
                mode_list = modes.get("lead") or modes.get(role) or modes.get(role.replace("_agent", ""))
                if isinstance(mode_list, list) and mode_list:
                    mode_set = {canonical_tool_name(str(n)) for n in mode_list if str(n).strip()}
                    mode_set = {n for n in mode_set if n}
                    read_only = mode_set & _LEAD_READ_TOOLS if mode_set else set(_LEAD_READ_TOOLS)
                    return (
                        read_only
                        | set(_LEAD_EXTRA_TOOLS)
                        | _mcp_allowlist_from_mapped(mapped)
                        | _lead_network_from_mapped(mapped)
                    )
        return _lead_allowed_tools(mapped)
    if norm_early in {"butler", "default", ""} or role == "butler":
        return None
    if project is None:
        return None
    raw = [canonical_tool_name(n) for n in (project.tools or [])]
    mapped = {n for n in raw if n}
    if not mapped:
        return None
    norm = role.replace("_agent", "").strip().lower()
    modes = getattr(project, "tool_modes", None) or {}
    if isinstance(modes, dict) and modes:
        mode_list = modes.get(norm) or modes.get(role) or modes.get(role.replace("_agent", ""))
        if isinstance(mode_list, list) and mode_list:
            mode_set = {canonical_tool_name(str(n)) for n in mode_list if str(n).strip()}
            mode_set = {n for n in mode_set if n}
            return mode_set if mode_set else None
    if norm == "dev":
        return mapped | set(dev_engine_extra_tools())
    return mapped


def _tool_allowed(name: str, allowed: set[str]) -> bool:
    if name in allowed:
        return True
    if "mcp_*" in allowed:

        if mcp_tool_allowed(name):
            return True
    return False


def filter_tool_definitions(
    tools: list[dict[str, Any]],
    allowed: set[str] | None,
) -> list[dict[str, Any]]:
    if allowed is None:
        return list(tools)
    return [
        t for t in tools
        if _tool_allowed(str(t.get("function", {}).get("name") or ""), allowed)
    ]


def _workflow_step_tool_allowlist(project: "Project | None") -> set[str] | None:

    return cast(set[str] | None, workflow_step_tool_allowlist(project))


def intersect_allowed_names(
    base: set[str] | None,
    extra: set[str] | None,
) -> set[str] | None:
    if extra is None:
        return base
    if base is None:
        return set(extra)
    return base & extra


def get_tool_definitions_for_project(
    project: "Project | None" = None,
    *,
    role: str = "butler",
    optimize_schema: bool = True,
) -> list[dict[str, Any]]:

    norm = role.replace("_agent", "").strip().lower()
    if norm in {"butler", "default", ""}:
        from butler.tools.registry import get_tool_definitions_unfiltered

        all_tools = get_tool_definitions_unfiltered()
    else:
        all_tools = get_tool_definitions()
    allowed = allowed_tool_names_for_project(project, role=role)
    allowed = intersect_allowed_names(allowed, _workflow_step_tool_allowlist(project))
    filtered = filter_tool_definitions(all_tools, allowed)
    if not optimize_schema:
        return filtered

    return cast(list[dict[str, Any]], optimize_tool_definitions_safe(filtered))


def get_current_project_tools(*, role: str = "butler") -> list[dict[str, Any]]:

    pm = get_project_manager()
    session_key = str(get_current_session_key() or "").strip()
    project = pm.get_current(session_key=session_key)
    return get_tool_definitions_for_project(project, role=role)


__all__ = [
    "allowed_tool_names_for_project",
    "canonical_tool_name",
    "filter_tool_definitions",
    "get_current_project_tools",
    "get_tool_definitions_for_project",
    "intersect_allowed_names",
]