"""The whitelist of mutations the assistant may propose.

Adding an entry here is the *only* way to widen the assistant's write surface,
and each entry owns four responsibilities that keep the model out of the
decision:

* ``schema`` — what the model is allowed to fill in, in tool-call form.
* ``normalise`` — server-side validation. Enum values are checked here, not
  trusted; anything the model gets wrong fails loudly at propose time rather
  than silently at execute time.
* ``summarise`` — the human-readable rows the employee reads before pressing
  confirm. If a parameter is not in this summary, the employee has not seen it,
  so every stored parameter must appear.
* ``execute`` — calls the domain's own service. No business logic lives here;
  the domain keeps its RBAC, idempotency, notification and approval behaviour
  exactly as the web form gets it.
"""

from __future__ import annotations

import re
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.core.locale import DEFAULT_LANG, Lang
from app.identity.person_models import IdentityPerson

ACTION_SUPPORT_TICKET = "create_support_ticket"
ACTION_VPN_APPLICATION = "submit_vpn_application"
ACTION_ASSET_REQUEST = "create_asset_request"
ACTION_COMPUTER_BUYBACK = "submit_computer_buyback"


@dataclass(frozen=True)
class SummaryRow:
    """One label/value line in the confirmation view."""

    label: dict[Lang, str]
    value: str

    def localized(self, lang: Lang) -> dict[str, str]:
        return {
            "label": self.label.get(lang) or self.label[DEFAULT_LANG],
            "value": self.value,
        }


@dataclass(frozen=True)
class AssistantAction:
    key: str
    title: dict[Lang, str]
    #: What the employee is told will happen, above the parameter rows.
    intent: dict[Lang, str]
    schema: dict[str, Any]
    normalise: Callable[[dict[str, Any]], dict[str, Any]]
    summarise: Callable[[dict[str, Any], Lang], list[SummaryRow]]
    execute: Callable[..., Awaitable[dict[str, Any]]]

    def localized_title(self, lang: Lang) -> str:
        return self.title.get(lang) or self.title[DEFAULT_LANG]

    def localized_intent(self, lang: Lang) -> str:
        return self.intent.get(lang) or self.intent[DEFAULT_LANG]


def _invalid(message: str) -> AppException:
    return AppException(
        code="ASSISTANT_ACTION_PARAMS_INVALID",
        message=message,
        status_code=422,
    )


#: A ``\n`` the model wrote as two characters instead of a newline. Not
#: preceded by another backslash, so an already-escaped ``\\n`` is left alone.
_LITERAL_NEWLINE = re.compile(r"(?<!\\)\\(?:r\\)?n")


def unescape_literal_newlines(text: str) -> str:
    """Turn a model's two-character ``\\n`` into an actual line break.

    Production, 2026-08-18: a confirmation card read
    ``项目名：abc\\n用途：bcd\\n预算：200 美元/月`` — backslash and n, visible, in
    the body the employee was being asked to approve. The model wrote ``\\\\n``
    inside its JSON tool arguments, so what arrived after decoding was the
    literal pair. Nothing on our side escapes anything; the fix belongs at the
    parameter boundary, before the value is stored and rendered.

    Applied ONLY to a string that contains no real newline — text the model
    formatted correctly is never rewritten. The residual risk is a Windows path
    (``C:\\new``) losing a backslash in free text; accepted knowingly, because
    the alternative is every structured card body carrying visible escapes, and
    a ticket body on this platform holds POSIX paths when it holds any.
    """
    if "\n" in text or "\\n" not in text:
        return text
    return _LITERAL_NEWLINE.sub("\n", text)


def _clean_text(value: Any, *, field: str, max_len: int, required: bool = True) -> str:
    text = unescape_literal_newlines(str(value or "")).strip()
    if required and not text:
        raise _invalid(f"{field} is required.")
    return text[:max_len]


def _label(table: dict[str, dict[Lang, str]], key: str, lang: Lang) -> str:
    """Human label for an enum value, in the reader's language.

    Falls back to the raw code rather than to Chinese: an unlabelled value is a
    gap in this table, and showing the code makes that visible instead of
    quietly rendering the wrong language at the reader.
    """
    entry = table.get(key)
    if not entry:
        return key
    return entry.get(lang) or entry.get(DEFAULT_LANG) or key


def _one_of(
    value: Any,
    allowed: tuple[str, ...],
    *,
    field: str,
    default: str | None = None,
) -> str:
    """Enum values are validated, never coerced silently.

    Two ways a summary could quietly become a lie, both closed here:

    * an *unknown* value — the domain services fall back to a default, which is
      right for a web form and wrong for something an employee is about to
      approve;
    * a *missing* value — a default filled in for a field the schema marks
      required would put a choice on the card that nobody made. Only fields
      with a genuinely safe default pass one; the rest must be asked for.
    """
    text = str(value or "").strip()
    if not text:
        if default is None:
            raise _invalid(f"{field} is required.")
        text = default
    if text not in allowed:
        raise _invalid(f"{field} must be one of {list(allowed)}; got {text!r}.")
    return text


# ── create_support_ticket ───────────────────────────────────────────

_TICKET_CATEGORIES = (
    "hardware",
    "account_access",
    "software",
    "network",
    "facility",
    "other",
)
_TICKET_PRIORITIES = ("low", "normal", "high", "urgent")

_CATEGORY_LABELS: dict[str, dict[Lang, str]] = {
    "hardware": {"zh": "硬件", "en": "Hardware", "ja": "ハードウェア"},
    "account_access": {"zh": "账号与权限", "en": "Account & access", "ja": "アカウント・権限"},
    "software": {"zh": "软件", "en": "Software", "ja": "ソフトウェア"},
    "network": {"zh": "网络", "en": "Network", "ja": "ネットワーク"},
    "facility": {"zh": "办公设施", "en": "Facility", "ja": "設備"},
    "other": {"zh": "其他", "en": "Other", "ja": "その他"},
}
_PRIORITY_LABELS: dict[str, dict[Lang, str]] = {
    "low": {"zh": "低", "en": "Low", "ja": "低"},
    "normal": {"zh": "普通", "en": "Normal", "ja": "通常"},
    "high": {"zh": "高", "en": "High", "ja": "高"},
    "urgent": {"zh": "紧急", "en": "Urgent", "ja": "緊急"},
}


def _normalise_ticket(params: dict[str, Any]) -> dict[str, Any]:
    return {
        "title": _clean_text(params.get("title"), field="title", max_len=200),
        "body": _clean_text(params.get("body"), field="body", max_len=4000),
        "category": _one_of(
            params.get("category"),
            _TICKET_CATEGORIES,
            field="category",
            default="other",
        ),
        "priority": _one_of(
            params.get("priority"),
            _TICKET_PRIORITIES,
            field="priority",
            default="normal",
        ),
    }


def _summarise_ticket(params: dict[str, Any], lang: Lang) -> list[SummaryRow]:
    category = str(params.get("category") or "other")
    priority = str(params.get("priority") or "normal")
    return [
        SummaryRow(
            {"zh": "标题", "en": "Title", "ja": "件名"},
            str(params.get("title") or ""),
        ),
        SummaryRow(
            {"zh": "问题描述", "en": "Description", "ja": "内容"},
            str(params.get("body") or ""),
        ),
        SummaryRow(
            {"zh": "分类", "en": "Category", "ja": "カテゴリ"},
            _label(_CATEGORY_LABELS, category, lang),
        ),
        SummaryRow(
            {"zh": "优先级", "en": "Priority", "ja": "優先度"},
            _label(_PRIORITY_LABELS, priority, lang),
        ),
    ]


async def _execute_ticket(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    params: dict[str, Any],
    proposal_id: uuid.UUID,
    trace_id: str,
) -> dict[str, Any]:
    """File the ticket through the support domain's own service.

    ``source="ai"`` so help-desk analytics can separate assistant-filed tickets
    from web ones. The service carries its own per-(requester, title) dedup
    window, which is a second belt under the proposal's single-use token.
    """
    from app.assistant.models import (  # noqa: PLC0415
        AssistantActionProposal,
        AssistantConversation,
    )
    from app.core.coordinators.assistant_support_coordinator import (  # noqa: PLC0415
        create_assistant_ticket,
    )

    proposal = await db.get(AssistantActionProposal, proposal_id)
    conversation = (
        await db.get(AssistantConversation, proposal.conversation_id)
        if proposal is not None and proposal.conversation_id is not None
        else None
    )

    ticket = await create_assistant_ticket(
        db,
        requester=person,
        title=str(params["title"]),
        body=str(params["body"]),
        category=str(params["category"]),
        priority=str(params["priority"]),
        service_request_id=(conversation.service_request_id if conversation else None),
        trace_id=trace_id,
    )
    return {
        "ticket_id": str(ticket.id),
        "ticket_number": ticket.ticket_number,
        "status": ticket.status,
    }


SUPPORT_TICKET_ACTION = AssistantAction(
    key=ACTION_SUPPORT_TICKET,
    title={
        "zh": "提交 IT 求助工单",
        "en": "Raise an IT support ticket",
        "ja": "IT サポート依頼を作成",
    },
    intent={
        "zh": "确认后将以你的名义提交一张 IT 求助工单，IT 会收到通知并跟进。",
        "en": (
            "Confirming files an IT support ticket in your name; IT is notified and will follow up."
        ),
        "ja": (
            "確認すると、あなたの名義で IT サポート依頼を作成します。"
            "IT に通知され、対応が始まります。"
        ),
    },
    schema={
        "type": "object",
        "properties": {
            "title": {
                "type": "string",
                "description": "One-line summary of the problem, in the employee's language.",
            },
            "body": {
                "type": "string",
                "description": (
                    "What is wrong, what was already tried, and anything IT needs "
                    "to reproduce it. Write it from the employee's point of view."
                ),
            },
            "category": {"type": "string", "enum": list(_TICKET_CATEGORIES)},
            "priority": {"type": "string", "enum": list(_TICKET_PRIORITIES)},
        },
        "required": ["title", "body"],
        "additionalProperties": False,
    },
    normalise=_normalise_ticket,
    summarise=_summarise_ticket,
    execute=_execute_ticket,
)


def get_action(key: str) -> AssistantAction | None:
    """A curated action if one exists, otherwise the tool derived from MCP.

    Curated wins so the flows with a hand-written confirmation card keep it.
    Everything else is still offerable with a generic card — the point of the
    fallback is that a missing card downgrades presentation, not availability.
    """
    curated = _REGISTRY.get(key)
    if curated is not None:
        return curated
    from app.assistant.derived_actions import get_derived  # noqa: PLC0415 — cycle

    return get_derived(key)


def all_actions() -> list[AssistantAction]:
    """Only the curated set — this is what the channel prompt lists by name.

    The derived set is reached through search, not enumeration: 150+ titles
    would crowd out the prompt and make the model choose worse, which is the
    same reason the read bridge is two-stage.
    """
    return list(_REGISTRY.values())


# ── submit_vpn_application ──────────────────────────────────────────

_VPN_NODES = ("CN4-HK", "CN5-US")

_VPN_NODE_LABELS: dict[str, dict[Lang, str]] = {
    "CN4-HK": {"zh": "CN4-HK（香港）", "en": "CN4-HK (Hong Kong)", "ja": "CN4-HK（香港）"},
    "CN5-US": {"zh": "CN5-US（美国）", "en": "CN5-US (United States)", "ja": "CN5-US（米国）"},
}


def _normalise_vpn(params: dict[str, Any]) -> dict[str, Any]:
    node = str(params.get("vpn_node") or "").strip().upper()
    if node not in _VPN_NODES:
        raise _invalid(f"vpn_node must be one of {list(_VPN_NODES)}; got {node!r}.")
    return {
        "vpn_node": node,
        "usage_description": _clean_text(
            params.get("usage_description"),
            field="usage_description",
            max_len=2000,
        ),
        "resource_checklist": _clean_text(
            params.get("resource_checklist"),
            field="resource_checklist",
            max_len=2000,
        ),
    }


def _summarise_vpn(params: dict[str, Any], lang: Lang) -> list[SummaryRow]:
    node = str(params.get("vpn_node") or "")
    return [
        SummaryRow(
            {"zh": "节点", "en": "Node", "ja": "ノード"},
            _label(_VPN_NODE_LABELS, node, lang),
        ),
        SummaryRow(
            {"zh": "用途说明", "en": "What it is for", "ja": "利用目的"},
            str(params.get("usage_description") or ""),
        ),
        SummaryRow(
            {"zh": "需要访问的资源", "en": "Resources needed", "ja": "アクセス対象"},
            str(params.get("resource_checklist") or ""),
        ),
    ]


async def _execute_vpn(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    params: dict[str, Any],
    proposal_id: uuid.UUID,
    trace_id: str,
) -> dict[str, Any]:
    """Submit through the VPN domain's own service.

    That service carries its own leaver gate and its own in-flight lock, both
    added for exactly this shape of caller (an agent reaching the service
    without passing the HTTP router). Nothing is re-implemented here.
    """
    from app.domains.vpn_application.service import (  # noqa: PLC0415
        VpnApplicationService,
    )

    application = await VpnApplicationService(db).submit(
        requester=person,
        vpn_node=str(params["vpn_node"]),
        usage_description=str(params["usage_description"]),
        resource_checklist=str(params["resource_checklist"]),
        trace_id=trace_id,
    )
    return {
        "application_id": str(application.id),
        "vpn_node": application.vpn_node,
        "status": application.status,
    }


VPN_APPLICATION_ACTION = AssistantAction(
    key=ACTION_VPN_APPLICATION,
    title={
        "zh": "申请 VPN Full Mode",
        "en": "Request VPN Full Mode",
        "ja": "VPN フルモードを申請",
    },
    intent={
        "zh": "确认后将以你的名义提交 VPN Full Mode 申请，走既有飞书审批。",
        "en": (
            "Confirming submits a VPN Full Mode application in your name, "
            "through the existing Feishu approval flow."
        ),
        "ja": (
            "確認すると、あなたの名義で VPN フルモード申請を送信します"
            "（既存の Feishu 承認フローを利用）。"
        ),
    },
    schema={
        "type": "object",
        "properties": {
            "vpn_node": {"type": "string", "enum": list(_VPN_NODES)},
            "usage_description": {
                "type": "string",
                "description": "Why the employee needs Full Mode, in their own words.",
            },
            "resource_checklist": {
                "type": "string",
                "description": "Which internal resources or hosts they need to reach.",
            },
        },
        "required": ["vpn_node", "usage_description", "resource_checklist"],
        "additionalProperties": False,
    },
    normalise=_normalise_vpn,
    summarise=_summarise_vpn,
    execute=_execute_vpn,
)


# ── create_asset_request ────────────────────────────────────────────

#: Deliberately a subset of the domain's request types. Borrowing and repair
#: are self-scoped and reversible; claim / handover / return move ownership
#: between people and belong on a screen with the asset in front of you.
_ASSET_REQUEST_TYPES = ("self_borrow", "repair")
_ASSET_CATEGORIES = (
    "laptop",
    "desktop",
    "monitor",
    "printer",
    "server",
    "network_device",
    "accessory",
    "conference_display",
)

_ASSET_TYPE_LABELS: dict[str, dict[Lang, str]] = {
    "self_borrow": {"zh": "借用设备", "en": "Borrow a device", "ja": "機器の貸出"},
    "repair": {"zh": "设备报修", "en": "Device repair", "ja": "機器の修理"},
}
_ASSET_CATEGORY_LABELS: dict[str, dict[Lang, str]] = {
    "laptop": {"zh": "笔记本", "en": "Laptop", "ja": "ノート PC"},
    "desktop": {"zh": "台式机", "en": "Desktop", "ja": "デスクトップ"},
    "monitor": {"zh": "显示器", "en": "Monitor", "ja": "モニター"},
    "printer": {"zh": "打印机", "en": "Printer", "ja": "プリンター"},
    "server": {"zh": "服务器", "en": "Server", "ja": "サーバー"},
    "network_device": {"zh": "网络设备", "en": "Network device", "ja": "ネットワーク機器"},
    "accessory": {"zh": "配件", "en": "Accessory", "ja": "アクセサリ"},
    "conference_display": {"zh": "会议屏", "en": "Conference display", "ja": "会議用ディスプレイ"},
}


def _normalise_asset_request(params: dict[str, Any]) -> dict[str, Any]:
    return {
        # No defaults: the schema marks both required, and inventing "borrow a
        # laptop" for a model that forgot to say would put a choice on the
        # confirmation card that nobody made.
        "request_type": _one_of(
            params.get("request_type"),
            _ASSET_REQUEST_TYPES,
            field="request_type",
        ),
        "category": _one_of(
            params.get("category"),
            _ASSET_CATEGORIES,
            field="category",
        ),
        "reason": _clean_text(params.get("reason"), field="reason", max_len=2000),
    }


def _summarise_asset_request(params: dict[str, Any], lang: Lang) -> list[SummaryRow]:
    request_type = str(params.get("request_type") or "")
    category = str(params.get("category") or "")
    return [
        SummaryRow(
            {"zh": "申请类型", "en": "Request type", "ja": "申請種別"},
            _label(_ASSET_TYPE_LABELS, request_type, lang),
        ),
        SummaryRow(
            {"zh": "设备类别", "en": "Device category", "ja": "機器カテゴリ"},
            _label(_ASSET_CATEGORY_LABELS, category, lang),
        ),
        SummaryRow(
            {"zh": "原因说明", "en": "Reason", "ja": "理由"},
            str(params.get("reason") or ""),
        ),
    ]


async def _execute_asset_request(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    params: dict[str, Any],
    proposal_id: uuid.UUID,
    trace_id: str,
) -> dict[str, Any]:
    """File the employee request through the asset domain's own service.

    That service has no dedup of its own — a repeated call opens a second
    request. The proposal's single-use token is therefore load-bearing here,
    not merely defence in depth.
    """
    from app.domains.asset.schemas import (  # noqa: PLC0415
        EmployeeRequestCreateRequest,
    )
    from app.domains.asset.service import AssetService  # noqa: PLC0415

    request = await AssetService(db).create_employee_request(
        EmployeeRequestCreateRequest(
            request_type=str(params["request_type"]),
            category=str(params["category"]),
            reason=str(params["reason"]),
        ),
        person.id,
    )
    await db.commit()
    return {
        "request_id": str(request.id),
        "request_number": request.request_number,
        "status": request.status,
    }


ASSET_REQUEST_ACTION = AssistantAction(
    key=ACTION_ASSET_REQUEST,
    title={
        "zh": "提交设备申请",
        "en": "Raise a device request",
        "ja": "機器申請を作成",
    },
    intent={
        "zh": "确认后将以你的名义提交设备申请，IT 收到后会处理。",
        "en": ("Confirming files a device request in your name for IT to process."),
        "ja": "確認すると、あなたの名義で機器申請を作成し、IT が対応します。",
    },
    schema={
        "type": "object",
        "properties": {
            "request_type": {"type": "string", "enum": list(_ASSET_REQUEST_TYPES)},
            "category": {"type": "string", "enum": list(_ASSET_CATEGORIES)},
            "reason": {
                "type": "string",
                "description": "Why the device is needed, or what is wrong with it.",
            },
        },
        "required": ["request_type", "category", "reason"],
        "additionalProperties": False,
    },
    normalise=_normalise_asset_request,
    summarise=_summarise_asset_request,
    execute=_execute_asset_request,
)


# ── submit_computer_buyback ─────────────────────────────────────────


def _normalise_computer_buyback(params: dict[str, Any]) -> dict[str, Any]:
    raw_asset_id = _clean_text(params.get("asset_id"), field="asset_id", max_len=64)
    try:
        asset_id = str(uuid.UUID(raw_asset_id))
    except (ValueError, TypeError) as exc:
        raise _invalid("asset_id must be the id of a device returned by Helix.") from exc
    return {
        "asset_id": asset_id,
        # This is a display-only corroboration partner for the opaque id: the
        # employee sees the device they selected on the confirmation card.
        "asset_name": _clean_text(params.get("asset_name"), field="asset_name", max_len=255),
        "reason": _clean_text(params.get("reason"), field="reason", max_len=1000),
    }


def _summarise_computer_buyback(params: dict[str, Any], _lang: Lang) -> list[SummaryRow]:
    return [
        SummaryRow(
            {"zh": "回购设备", "en": "Device to repurchase", "ja": "買い取り対象端末"},
            str(params["asset_name"]),
        ),
        SummaryRow(
            {"zh": "设备标识", "en": "Device identifier", "ja": "端末識別子"},
            str(params["asset_id"]),
        ),
        SummaryRow(
            {"zh": "申请说明", "en": "Request note", "ja": "申請内容"}, str(params["reason"])
        ),
    ]


async def _execute_computer_buyback(
    _db: AsyncSession,
    *,
    person: IdentityPerson,
    params: dict[str, Any],
    proposal_id: uuid.UUID,
    trace_id: str,
) -> dict[str, Any]:
    """Submit the existing offboarding-buyback approval through its MCP twin.

    This deliberately does not reimplement pricing, eligibility, approval or
    idempotency.  The approval tool derives the price from the selected asset
    under the policy and carries the same guards as the employee web form.
    """
    from app.mcp.context import mcp_trace_id, mcp_user_id  # noqa: PLC0415
    from app.mcp.server import mcp_server  # noqa: PLC0415

    tool = mcp_server._tool_manager._tools.get("submit_approval_request")  # noqa: SLF001
    if tool is None:
        raise RuntimeError("submit_approval_request is not registered")
    user_token = mcp_user_id.set(person.id)
    trace_token = mcp_trace_id.set(uuid.UUID(str(trace_id)))
    try:
        result = await tool.run(
            {
                "template_key": "offboarding_buyback",
                "asset_id": params["asset_id"],
                "reason": params["reason"],
                "idempotency_key": f"assistant-{proposal_id}",
            }
        )
    finally:
        mcp_user_id.reset(user_token)
        mcp_trace_id.reset(trace_token)
    # Same reasoning as the derived wrapper: ``convert_result=False`` means
    # ``result`` is the tool's own return value, and for this one that is the
    # approval-backed request response carrying the ``id`` the lifecycle needs.
    # Stringifying it here would leave buyback as the one flow that submits
    # successfully and can never be followed up on.
    from app.assistant.derived_actions import _payload  # noqa: PLC0415 — cycle

    return {"tool": "submit_approval_request", "result": _payload(result)}


COMPUTER_BUYBACK_ACTION = AssistantAction(
    key=ACTION_COMPUTER_BUYBACK,
    title={
        "zh": "发起电脑回购申请",
        "en": "Start a computer buyback request",
        "ja": "PC 買い取り申請を開始",
    },
    intent={
        "zh": "确认后将以你的名义发起电脑回购审批；回购价格与资格由 Helix 按所选设备的公司政策计算。",
        "en": "Confirming starts the computer buyback approval. Helix calculates eligibility and price from the selected device under company policy.",
        "ja": "確認すると、選択した端末に対する買い取り承認を開始します。価格と対象可否は Helix が会社ポリシーに基づき計算します。",
    },
    schema={
        "type": "object",
        "properties": {
            "asset_id": {
                "type": "string",
                "description": "ID of the employee-owned device returned by a Helix lookup. Never invent it.",
            },
            "asset_name": {
                "type": "string",
                "description": "The matching device name or asset code returned with asset_id.",
            },
            "reason": {
                "type": "string",
                "description": "Why the employee is requesting the buyback, in their own words.",
            },
        },
        "required": ["asset_id", "asset_name", "reason"],
        "additionalProperties": False,
    },
    normalise=_normalise_computer_buyback,
    summarise=_summarise_computer_buyback,
    execute=_execute_computer_buyback,
)


_REGISTRY: dict[str, AssistantAction] = {
    action.key: action
    for action in (
        SUPPORT_TICKET_ACTION,
        VPN_APPLICATION_ACTION,
        ASSET_REQUEST_ACTION,
        COMPUTER_BUYBACK_ACTION,
    )
}