content-find-agent-new/content_agent/errors.py

from __future__ import annotations

from enum import StrEnum
from typing import Any


class ErrorCode(StrEnum):
    RUN_START_FAILED = "RUN_START_FAILED"
    RUN_NOT_FOUND = "RUN_NOT_FOUND"
    INVALID_REQUEST = "INVALID_REQUEST"
    INVALID_SOURCE = "INVALID_SOURCE"
    DB_CONFIG_MISSING = "DB_CONFIG_MISSING"
    DB_CONNECTION_FAILED = "DB_CONNECTION_FAILED"
    DB_PERMISSION_DENIED = "DB_PERMISSION_DENIED"
    DB_SCHEMA_NOT_READY = "DB_SCHEMA_NOT_READY"
    RUNTIME_WRITE_FAILED = "RUNTIME_WRITE_FAILED"
    POLICY_BUNDLE_NOT_FOUND = "POLICY_BUNDLE_NOT_FOUND"
    PLATFORM_CONFIG_MISSING = "PLATFORM_CONFIG_MISSING"
    PLATFORM_REQUEST_FAILED = "PLATFORM_REQUEST_FAILED"
    QUERY_GENERATION_FAILED = "QUERY_GENERATION_FAILED"
    CONFIG_RULE_PACK_DISPATCH_CONFLICT = "CONFIG_RULE_PACK_DISPATCH_CONFLICT"


SENSITIVE_KEYS = {
    "password",
    "token",
    "access_token",
    "refresh_token",
    "api_key",
    "apikey",
    "secret",
    "dsn",
    "authorization",
    "cookie",
    "session",
    "credential",
}


class ContentAgentError(Exception):
    def __init__(
        self,
        error_code: ErrorCode | str,
        message: str,
        detail: dict[str, Any] | None = None,
        status_code: int = 500,
    ) -> None:
        super().__init__(message)
        self.error_code = ErrorCode(error_code)
        self.message = message
        self.detail = sanitize_error_detail(detail or {})
        self.status_code = status_code

    def response_detail(self) -> dict[str, Any]:
        return error_response(self.error_code, self.message, self.detail)


def error_response(
    error_code: ErrorCode | str,
    message: str,
    detail: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return {
        "error_code": ErrorCode(error_code).value,
        "message": message,
        "detail": sanitize_error_detail(detail or {}),
    }


def sanitize_error_detail(value: Any) -> Any:
    if isinstance(value, dict):
        result: dict[str, Any] = {}
        for key, item in value.items():
            lowered = str(key).lower()
            if lowered in SENSITIVE_KEYS:
                result[key] = "<redacted>"
            else:
                result[key] = sanitize_error_detail(item)
        return result
    if isinstance(value, list):
        return [sanitize_error_detail(item) for item in value]
    if isinstance(value, tuple):
        return [sanitize_error_detail(item) for item in value]
    return value


def error_from_exception(
    exc: Exception,
    *,
    default_code: ErrorCode = ErrorCode.RUN_START_FAILED,
    detail: dict[str, Any] | None = None,
) -> ContentAgentError:
    if isinstance(exc, ContentAgentError):
        return exc
    return ContentAgentError(
        default_code,
        _safe_message(default_code),
        {
            **(detail or {}),
            "exception_type": type(exc).__name__,
        },
    )


def _safe_message(error_code: ErrorCode) -> str:
    messages = {
        ErrorCode.RUN_START_FAILED: "run start failed",
        ErrorCode.RUNTIME_WRITE_FAILED: "runtime write failed",
        ErrorCode.INVALID_SOURCE: "invalid source",
        ErrorCode.INVALID_REQUEST: "invalid request",
        ErrorCode.POLICY_BUNDLE_NOT_FOUND: "policy bundle not found",
        ErrorCode.PLATFORM_CONFIG_MISSING: "platform config missing",
        ErrorCode.PLATFORM_REQUEST_FAILED: "platform request failed",
        ErrorCode.QUERY_GENERATION_FAILED: "query generation failed",
        ErrorCode.CONFIG_RULE_PACK_DISPATCH_CONFLICT: "rule pack dispatch conflict in config",
    }
    return messages.get(error_code, error_code.value.lower())