ForcePilot/backend/package/yuxi/channels/contract/errors/client.py

"""客户端错误类型定义。

定义 ``ClientError`` 抽象基类及其 4 个具体子类，表示由调用方引起的错误
（4xx 语义）：参数校验失败、认证失败、资源不存在、权限不足。
"""

from __future__ import annotations

from typing import Any

from yuxi.channels.contract.errors.base import Error


class ClientError(Error):
    """客户端错误抽象基类。

    表示由调用方引起的错误（4xx 语义）。子类包括 ``ValidationError`` /
    ``AuthError`` / ``NotFoundError`` / ``PermissionDeniedError``。
    """

    error_code = "CLIENT_ERROR"


class ValidationError(ClientError):
    """字段校验失败错误。"""

    error_code = "VALIDATION_ERROR"

    def __init__(
        self,
        field: str,
        message: str,
        *,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(message, trace_id=trace_id)
        self.field = field

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["field"] = self.field
        return data


class AuthError(ClientError):
    """认证失败错误。

    调用方身份认证失败时抛出，如 token 无效、签名错误（HTTP 401）。
    """

    error_code = "AUTH_ERROR"

    def __init__(
        self,
        reason: str,
        *,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(f"Authentication failed: {reason}", trace_id=trace_id)
        self.reason = reason

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["reason"] = self.reason
        return data


class NotFoundError(ClientError):
    """资源不存在错误。

    调用方请求的资源不存在时抛出（HTTP 404）。
    """

    error_code = "NOT_FOUND"

    def __init__(
        self,
        resource: str,
        id: str,
        *,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(f"{resource} not found: {id}", trace_id=trace_id)
        self.resource = resource
        self.id = id

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["resource"] = self.resource
        data["id"] = self.id
        return data


class PermissionDeniedError(ClientError):
    """权限不足错误。

    调用方无权限执行请求的操作时抛出（HTTP 403）。命名为 ``PermissionDeniedError``
    以避免与 Python 内置 ``PermissionError`` 冲突。
    """

    error_code = "PERMISSION_DENIED"

    def __init__(
        self,
        reason: str,
        *,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(f"Permission denied: {reason}", trace_id=trace_id)
        self.reason = reason

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["reason"] = self.reason
        return data


class ServiceAccountNotFoundError(NotFoundError):
    """渠道账户未绑定服务账号错误。

    渠道账户未关联服务账号（``User.user_type='service'``）时抛出（HTTP 404）。
    ``details`` 包含 ``channel_type`` 与 ``account_id``，便于定位缺失绑定的
    渠道账户。
    """

    error_code = "CHANNEL_SERVICE_ACCOUNT_NOT_FOUND"

    def __init__(
        self,
        channel_type: str,
        account_id: str,
        *,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(
            "service_account",
            f"{channel_type}:{account_id}",
            trace_id=trace_id,
        )
        self.channel_type = channel_type
        self.account_id = account_id

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["channel_type"] = self.channel_type
        data["account_id"] = self.account_id
        return data


class ChannelAccessDeniedError(PermissionDeniedError):
    """渠道访客无权访问 Agent 错误。

    渠道访客在当前渠道访问级别下无权访问目标 Agent 时抛出（HTTP 403）。
    ``details`` 包含 ``agent_slug``、``channel_access_level``，可选携带
    ``unified_identity_id`` 用于跨渠道身份关联排查。
    """

    error_code = "CHANNEL_ACCESS_DENIED"

    def __init__(
        self,
        agent_slug: str,
        channel_access_level: str,
        *,
        unified_identity_id: str | None = None,
        trace_id: str | None = None,
    ) -> None:
        super().__init__(
            f"access denied to agent '{agent_slug}' at level '{channel_access_level}'",
            trace_id=trace_id,
        )
        self.agent_slug = agent_slug
        self.channel_access_level = channel_access_level
        self.unified_identity_id = unified_identity_id

    def to_dict(self) -> dict[str, Any]:
        data = super().to_dict()
        data["agent_slug"] = self.agent_slug
        data["channel_access_level"] = self.channel_access_level
        if self.unified_identity_id is not None:
            data["unified_identity_id"] = self.unified_identity_id
        return data