FusionAGI/fusionagi/api/error_codes.py

"""Structured error codes for machine-readable error taxonomy.

Every API error includes a unique code, human-readable message,
and optional details for programmatic handling.
"""

from __future__ import annotations

from enum import Enum
from typing import Any


class ErrorCode(str, Enum):
    """Machine-readable error codes for the FusionAGI API."""

    # Auth errors (1xxx)
    AUTH_MISSING = "FAGI-1001"
    AUTH_INVALID = "FAGI-1002"
    AUTH_EXPIRED = "FAGI-1003"
    AUTH_INSUFFICIENT = "FAGI-1004"

    # Rate limiting (2xxx)
    RATE_LIMIT_IP = "FAGI-2001"
    RATE_LIMIT_TENANT = "FAGI-2002"

    # Session errors (3xxx)
    SESSION_NOT_FOUND = "FAGI-3001"
    SESSION_EXPIRED = "FAGI-3002"
    SESSION_LIMIT = "FAGI-3003"

    # Prompt/input errors (4xxx)
    PROMPT_EMPTY = "FAGI-4001"
    PROMPT_TOO_LONG = "FAGI-4002"
    INPUT_INVALID = "FAGI-4003"
    FILE_TOO_LARGE = "FAGI-4004"

    # Orchestration errors (5xxx)
    ORCHESTRATOR_UNAVAILABLE = "FAGI-5001"
    HEAD_TIMEOUT = "FAGI-5002"
    WITNESS_FAILURE = "FAGI-5003"
    CONSENSUS_FAILURE = "FAGI-5004"

    # Adapter errors (6xxx)
    LLM_UNAVAILABLE = "FAGI-6001"
    LLM_TIMEOUT = "FAGI-6002"
    LLM_RATE_LIMIT = "FAGI-6003"
    LLM_CONTEXT_LENGTH = "FAGI-6004"

    # Governance errors (7xxx)
    GOVERNANCE_ADVISORY = "FAGI-7001"
    SAFETY_FLAG = "FAGI-7002"
    PII_DETECTED = "FAGI-7003"

    # Infrastructure errors (8xxx)
    DB_UNAVAILABLE = "FAGI-8001"
    CACHE_UNAVAILABLE = "FAGI-8002"
    STORAGE_FULL = "FAGI-8003"

    # Tenant errors (9xxx)
    TENANT_NOT_FOUND = "FAGI-9001"
    TENANT_SUSPENDED = "FAGI-9002"

    # General (0xxx)
    INTERNAL_ERROR = "FAGI-0001"
    NOT_IMPLEMENTED = "FAGI-0002"
    VERSION_UNSUPPORTED = "FAGI-0003"


# Human-readable descriptions
_DESCRIPTIONS: dict[ErrorCode, str] = {
    ErrorCode.AUTH_MISSING: "Authentication required. Provide a Bearer token.",
    ErrorCode.AUTH_INVALID: "Invalid API key or token.",
    ErrorCode.AUTH_EXPIRED: "API key has expired. Rotate via /v1/admin/keys/rotate.",
    ErrorCode.AUTH_INSUFFICIENT: "Insufficient permissions for this operation.",
    ErrorCode.RATE_LIMIT_IP: "IP-level rate limit exceeded.",
    ErrorCode.RATE_LIMIT_TENANT: "Tenant-level rate limit exceeded.",
    ErrorCode.SESSION_NOT_FOUND: "Session not found. Create one via POST /v1/sessions.",
    ErrorCode.SESSION_EXPIRED: "Session has expired.",
    ErrorCode.SESSION_LIMIT: "Maximum concurrent sessions reached.",
    ErrorCode.PROMPT_EMPTY: "Prompt cannot be empty.",
    ErrorCode.PROMPT_TOO_LONG: "Prompt exceeds maximum length.",
    ErrorCode.INPUT_INVALID: "Request body validation failed.",
    ErrorCode.FILE_TOO_LARGE: "Uploaded file exceeds size limit.",
    ErrorCode.ORCHESTRATOR_UNAVAILABLE: "Orchestrator is not initialized.",
    ErrorCode.HEAD_TIMEOUT: "One or more heads timed out during processing.",
    ErrorCode.WITNESS_FAILURE: "Witness synthesis failed.",
    ErrorCode.CONSENSUS_FAILURE: "Head consensus could not be reached.",
    ErrorCode.LLM_UNAVAILABLE: "LLM provider is unavailable.",
    ErrorCode.LLM_TIMEOUT: "LLM request timed out.",
    ErrorCode.LLM_RATE_LIMIT: "LLM provider rate limit hit.",
    ErrorCode.LLM_CONTEXT_LENGTH: "Input exceeds LLM context window.",
    ErrorCode.GOVERNANCE_ADVISORY: "Governance advisory triggered.",
    ErrorCode.SAFETY_FLAG: "Safety pipeline flagged the output.",
    ErrorCode.PII_DETECTED: "Potential PII detected in output.",
    ErrorCode.DB_UNAVAILABLE: "Database backend is unavailable.",
    ErrorCode.CACHE_UNAVAILABLE: "Cache backend is unavailable.",
    ErrorCode.STORAGE_FULL: "Storage capacity reached.",
    ErrorCode.TENANT_NOT_FOUND: "Tenant not found.",
    ErrorCode.TENANT_SUSPENDED: "Tenant account is suspended.",
    ErrorCode.INTERNAL_ERROR: "An unexpected internal error occurred.",
    ErrorCode.NOT_IMPLEMENTED: "This feature is not yet implemented.",
    ErrorCode.VERSION_UNSUPPORTED: "Requested API version is not supported.",
}


def error_response(
    code: ErrorCode,
    detail: str | None = None,
    extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a structured error response dict.

    Args:
        code: ErrorCode enum value.
        detail: Optional human-readable detail (overrides default).
        extra: Optional additional context.

    Returns:
        Structured error dict with code, message, and optional details.
    """
    resp: dict[str, Any] = {
        "error": {
            "code": code.value,
            "message": detail or _DESCRIPTIONS.get(code, "Unknown error"),
        },
    }
    if extra:
        resp["error"]["details"] = extra
    return resp


def error_json_response(
    code: ErrorCode,
    status_code: int = 400,
    detail: str | None = None,
    extra: dict[str, Any] | None = None,
) -> Any:
    """Build a FastAPI JSONResponse with structured error.

    Args:
        code: ErrorCode enum value.
        status_code: HTTP status code.
        detail: Optional override message.
        extra: Optional additional context.

    Returns:
        JSONResponse with structured error body.
    """
    from starlette.responses import JSONResponse

    return JSONResponse(
        content=error_response(code, detail, extra),
        status_code=status_code,
    )