errors.py

"""Exception taxonomy (BG §1.2).

Mapping gateway policy rejections to catchable, actionable exceptions is the
SDK's clearest value over raw HTTP.

Two design points that matter:

1. :class:`PolicyViolation` must be distinguishable from a transient error at the framework
   boundary so host frameworks do not silently retry a refusal. It is NEVER retried by our
   transport.
2. ``remediation`` is a structurally-guaranteed, human-readable next step — worth more than a
   stack trace. Every :class:`DonkeyError` carries one: the base constructor refuses to build
   an instance whose remediation is empty or whitespace, and every subclass ships a canonical
   default (BG §1.2, #182, #715). That default is the single source ``donkey doctor`` (#202)
   reuses, so a diagnosis and the exception it stands for can never disagree.

The concrete HTTP-response → exception mapping lives in :func:`classify`: an ordered series of
discriminators, each keyed on a rejection shape captured from a real gateway
(docs/verified-apis.md §4, ``src/donkey_kit/simulator/_fixtures/anypoint/llm_proxy/``, BG §1.5)
rather than a hand-written guess. The CIBA step-up types live in ``core/step_up`` (#1320). A
shape it does not recognise falls through to a generic :class:`PolicyViolation` or
:class:`DonkeyError`.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from . import _wire
from ._refusal_body import (
    MODEL_NOT_ROUTABLE_RE,
    code_str,
    content_safety_reject,
    json_body,
    nested_error,
    parse_retry_after,
    pii_spans,
    pii_summary,
    sent_ids,
    str_or_none,
)
from ._refusal_body import retry_after as read_retry_after
from ._response import RequestLike, ResponseLike
from .lastcall import request_id as read_request_id

if TYPE_CHECKING:
    from datetime import datetime

__all__ = [
    "AgentKilled",
    "AuthError",
    "BudgetReserveReached",
    "ConfigError",
    "ConfigWarning",
    "ContentSafetyBlocked",
    "DonkeyError",
    "GatewayUnavailable",
    "ModelNotRoutable",
    "ModelSubstituted",
    "PIIDetected",
    "PolicyViolation",
    "PromptInjectionBlocked",
    "PublicationDrift",
    "RegistryError",
    "RequestLike",
    "RequestRateLimitExceeded",
    "ResponseLike",
    "TokenBudgetExceeded",
    "ToolInvocationError",
    "UpstreamModelError",
    "UpstreamRequestError",
    "classify",
    "gateway_unavailable",
    "parse_retry_after",
]


class DonkeyError(Exception):
    """Base for all SDK errors. Carries correlation/call/request IDs and the raw
    response so callers can inspect what actually happened.

    Three ids, three provenances (BG §1.1, #195):

    * ``correlation_id`` — the RUN id the client sent (``X-Correlation-Id``),
      shared by every request in a ``donkey.run()`` block. This is the
      client↔gateway join key, and it always equals the header that was sent.
    * ``call_id`` — the per-request id the client sent (``X-Donkey-Request-Id``), unique per
      logical request, stable across its retries. It pinpoints one request within a run and
      exists even when the request fails before any response.
    * ``request_id`` — the UPSTREAM PROVIDER's own id, passed through by the
      gateway and read back from a RESPONSE header whose name varies by provider
      (``x-request-id`` / ``x-amzn-requestid`` / ``apim-request-id``, #542). Quote
      it to the provider's support team. Absent on a transport error (no response)
      — and on any route where the provider forwarded none — unlike the two
      client-sent ids above. The gateway-side join key is ``correlation_id``.

    ``remediation`` names the caller's next step and is **structurally mandatory** on every
    error in the tree (#182, #715), so a handler written as ``except DonkeyError as e:
    log(e.remediation)`` never fails inside its own ``except``. An explicit ``remediation=``
    wins; otherwise the most-derived class default applies. The constructor raises
    :class:`ValueError` if the resolved value is empty or whitespace.
    """

    #: The framework exception this error was mapped from, if any. Kept here
    #: rather than chained as ``__cause__`` because its message can repeat
    #: request content, and a traceback renders every chained exception.
    framework_error: BaseException | None = None

    #: Default next-step wording. This base value covers the generic
    #: ``DonkeyError`` :func:`classify` returns for a response it cannot type;
    #: every subclass overrides it.
    remediation: str = (
        "The SDK could not map this failure to a more specific error. Inspect "
        ".response (status, headers, body) and quote .correlation_id and "
        ".request_id when reporting it."
    )

    def __init__(
        self,
        message: str,
        *,
        remediation: str | None = None,
        correlation_id: str | None = None,
        call_id: str | None = None,
        request_id: str | None = None,
        response: ResponseLike | None = None,
    ) -> None:
        super().__init__(message)
        resolved = remediation if remediation is not None else type(self).remediation
        if not resolved.strip():
            raise ValueError(
                f"{type(self).__name__} requires a non-empty remediation naming the "
                "caller's next step (BG §1.2, #182)."
            )
        self.remediation = resolved
        self.correlation_id = correlation_id
        self.call_id = call_id
        self.request_id = request_id
        self.response = response


class ConfigError(DonkeyError):
    """Configuration is missing or invalid, or an endpoint may not receive the
    credentials that would be sent to it (see ``DonkeyConfig.check_endpoints``).
    Raised locally before any request; reports ALL missing fields at once
    (config resolution). The transport also raises it, with its own
    remediation, for a send on a closed client or from a closed event loop
    (#813)."""

    remediation: str = (
        "Fix the configuration the message names — set each missing or invalid "
        "field as a keyword argument, a DONKEY_* environment variable, or in "
        ".donkey-kit.toml — then re-run. `donkey doctor` checks the resolved "
        "configuration."
    )


class ConfigWarning(UserWarning):
    """A setting accepted but likely a mistake: a misplaced config entry, an env switch that relaxes
    a check, a non-loopback ``serve()`` host, or a ``wait_for_gateway_approval()`` over 2 min."""


class AuthError(DonkeyError):
    """Rejected credentials on the Anypoint control plane or LLM-proxy data plane.

    The class-level ``remediation`` is the LLM-proxy data-plane default that
    ``donkey doctor`` (#202) reuses. Control-plane callers override it with one
    of the canonical class values below, so the next step matches the auth
    provider that failed (#484)."""

    #: Single source of next-step wording for a rejected-credentials diagnosis.
    remediation: str = (
        "The gateway rejected the credentials (401/403). Check "
        "DONKEY_LLM_PROXY_CLIENT_ID / DONKEY_LLM_PROXY_CLIENT_SECRET are the "
        "consumer client_id/secret pair for this LLM-proxy instance — not an "
        "Anypoint control-plane credential and not a bearer token — and that the "
        "consumer is authorized on the instance in API Manager (docs/verified-apis.md §2)."
    )
    connected_app_remediation: str = (
        "Check the configured Anypoint control-plane auth provider. For a connected app, "
        "verify ANYPOINT_CLIENT_ID / ANYPOINT_CLIENT_SECRET and that the app has the scopes "
        "the operation needs (docs/scopes.md, https://docs.donkey-kit.dev/reference/scopes)."
    )
    provider_chain_remediation: str = (
        "Every configured Anypoint control-plane auth provider failed. Inspect the last "
        "error and verify each provider's credentials or token source. If the chain uses "
        "a connected app, also verify its scopes (https://docs.donkey-kit.dev/reference/scopes)."
    )


class PolicyViolation(DonkeyError):
    """Base for gateway-enforced refusals. NEVER retried.

    ``remediation`` names the concrete next step the caller can take, e.g. "Token budget
    exceeded for business group `finance`; limit resets in 42m; request an increase in API
    Manager". It is **structurally mandatory** (#182), enforced by the :class:`DonkeyError`
    constructor, because a typed refusal with no next step is just a renamed exception. When
    no ``remediation`` is passed, the class-level :attr:`remediation` default applies; every
    concrete subclass ships its own, and that default is the single source ``donkey doctor``
    (#202) reuses so the CLI and the exception never disagree. It names the *action*, not the
    policy that fired.
    """

    policy: str = "unknown"

    #: Default next-step wording, used when the caller passes no ``remediation``.
    #: This base value is the generic-refusal fall-through (the shape ``classify``
    #: cannot pin more precisely); every concrete subclass overrides it.
    remediation: str = (
        "A gateway policy refused this request. This is terminal and was NOT retried. PII (403), "
        "agent kill switch (403, error code agent_killed), request-rate-limit (429 with "
        "x-ratelimit-* headers), token-budget (429), prompt-injection (the x-injection-protection "
        "header or the regex prompt guard's matched_patterns) and content-safety (Azure Content "
        "Safety, Amazon Bedrock Guardrails, Google Model Armor; CrowdStrike AIDR from its policy "
        "source) rejections have their own types; only federated-guardrail verdicts and otherwise-"
        "unrecognised shapes fall through to here. Inspect .response for the raw body."
    )

    def __init__(self, message: str, *, policy: str | None = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        if policy is not None:
            self.policy = policy


class TokenBudgetExceeded(PolicyViolation):
    """A token-rate-limit policy refused the request: the budget window is spent.

    Raised on the gateway's 429 token-budget rejection (BG §1.3). Terminal and
    never retried by the SDK, like every :class:`PolicyViolation`.

    Args:
        message: The human-readable refusal text.
        retry_after: Seconds until the caller may retry, read from the
            ``retry-after`` or ``x-token-reset`` response header; ``None`` when
            the gateway sent neither.

    To pace calls before hitting the limit, use ``donkey.budget`` and
    :class:`BudgetReserveReached` instead.

    Docs: https://docs.donkey-kit.dev/errors
    """

    policy = "token-rate-limit"
    remediation: str = (
        "A token-rate-limit policy exhausted the budget window. Wait for it to reset (see "
        "retry_after / x-token-reset) or request an increase in API Manager."
    )

    def __init__(self, message: str, *, retry_after: float | None = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        self.retry_after = retry_after


class RequestRateLimitExceeded(PolicyViolation):
    """A request-rate-limit policy refused the request: the window's request
    count is spent (docs/verified-apis.md §4, #974).

    The stock ``rate-limiting`` policy counts requests, not tokens. Its 429 has the body
    ``{"error":"Too Many Requests"}`` and the ``x-ratelimit-limit``, ``x-ratelimit-remaining``
    and ``x-ratelimit-reset`` headers, and no ``retry-after``. Terminal: never retried by the
    SDK, like every :class:`PolicyViolation`.

    Args:
        message: The human-readable refusal text.
        retry_after: Seconds until the window resets, read from the
            ``retry-after`` or ``x-ratelimit-reset`` response header; ``None``
            when the gateway sent neither.

    To pace calls before hitting the limit, use ``donkey.budget.requests`` and
    :class:`BudgetReserveReached` instead.

    Docs: https://docs.donkey-kit.dev/errors
    """

    policy = "request-rate-limit"
    remediation: str = (
        "A request-rate-limit policy allows no more requests in this window. "
        "Wait for it to reset (see retry_after / x-ratelimit-reset), send fewer "
        "requests per window (donkey.budget.pace() refuses before the limit), or "
        "request a higher limit in API Manager."
    )

    def __init__(self, message: str, *, retry_after: float | None = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        self.retry_after = retry_after


class BudgetReserveReached(DonkeyError):
    """Raised by :meth:`Budget.pace` *before* it lets a request through that would
    cross the caller's reserve (BG §1.3, #186).

    Deliberately NOT a :class:`PolicyViolation`: that base marks a gateway-enforced
    refusal that is terminal and never retried. This is the client-side opposite —
    a pre-emptive signal, raised locally from observed budget headers before any
    request is sent. When :attr:`reset_at` is known, the caller can recover with
    ``await budget.wait_for_reset()`` then continue. When it is ``None``, waiting
    cannot make progress, so the caller must handle or propagate the signal instead
    of retrying immediately. Classifying it as a refusal would misrepresent it at
    the framework boundary.

    Carries the observed budget state at the moment pacing tripped so a handler can
    decide without re-reading the object: ``window``, which window tripped
    (``"tokens"`` or ``"requests"``, #974), its ``fraction_used`` (0.0-1.0), the
    ``reserve`` that was requested, and its ``reset_at`` (``None`` if the reset
    time has not been observed)."""

    remediation: str = (
        "When reset_at is known, call await budget.wait_for_reset() before retrying. "
        "When reset_at is None, waiting cannot make progress; preserve the last "
        "checkpoint and handle or propagate this exception instead of retrying "
        "immediately."
    )

    def __init__(
        self,
        message: str,
        *,
        fraction_used: float,
        reserve: float,
        reset_at: datetime | None = None,
        window: str = "tokens",
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.fraction_used = fraction_used
        self.reserve = reserve
        self.reset_at = reset_at
        self.window = window


class ModelSubstituted(DonkeyError):
    """The gateway served a *different* model than the one requested — a routing
    fallback substituted the model, and ``on_model_substitution="raise"`` opted
    the caller into treating that as an error (docs/verified-apis.md §3, #309).

    Deliberately NOT a :class:`PolicyViolation`: the request was neither refused
    nor failed — it succeeded, just against a model the developer did not choose.
    It is a client-side determinism signal (like :class:`BudgetReserveReached`),
    off by default; a caller who has not opted in gets the substitution surfaced
    passively on ``donkey.last_call.substituted`` instead. Carries the response,
    so a handler can still read the served completion if it decides to accept it.

    ``request_id`` (the gateway's own id) is inherited from :class:`DonkeyError`
    and populated from the response, so a substitution can be quoted in a ticket
    on the same terms as any other gateway event."""

    remediation: str = (
        "The gateway served a different model than requested (a routing "
        "fallback). Accept the completion on .response, retry once the requested "
        "model is available, or leave on_model_substitution unset to observe "
        "substitutions on donkey.last_call.substituted instead of raising."
    )

    def __init__(
        self,
        message: str,
        *,
        requested_model: str,
        served_model: str,
        served_provider: str | None = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.requested_model = requested_model
        self.served_model = served_model
        self.served_provider = served_provider


class PromptInjectionBlocked(PolicyViolation):
    """The prompt-injection-protection policy flagged the request as an injection attempt.

    Identified from the ``x-injection-protection`` header or the regex prompt
    guard's ``matched_patterns``. Terminal and never retried by the SDK, like
    every :class:`PolicyViolation`. The raw body is on ``.response``.

    Docs: https://docs.donkey-kit.dev/errors
    """

    policy = "prompt-injection-protection"
    remediation: str = (
        "The prompt-injection-protection policy flagged this request as a "
        "prompt-injection attempt. Review and sanitise the untrusted input "
        "in the prompt, or adjust the policy's sensitivity in API Manager."
    )


class ContentSafetyBlocked(PolicyViolation):
    """A provider-backed content-moderation policy (Azure Content Safety, Amazon Bedrock
    Guardrails, Google Model Armor or CrowdStrike AIDR) blocked the request or response. The
    AIDR shape is read from the policy's source, not captured live (docs/verified-apis.md §4).

    ``provider`` is the vendor's stable slug (``azure-content-safety``, ``bedrock-guardrails``,
    ``google-model-armor``, ``crowdstrike-aidr``); ``categories`` the flagged reasons from its
    ``x-llm-proxy-<vendor>-...-reason`` header (e.g. ``severity_hate``, ``denied_topic``), the
    moderation analog of :attr:`PIIDetected.entities`, empty when none was reported. Both ride
    the span as ``donkey.policy.provider`` / ``donkey.policy.categories`` (#1183)."""

    policy = "content-safety"
    remediation: str = (
        "A content-moderation policy (Azure Content Safety, Bedrock Guardrails, Model Armor or "
        "CrowdStrike AIDR) blocked this request. Revise the flagged content, or adjust its "
        "categories / thresholds (API Manager; Model Armor template; AIDR policy in Falcon). "
        "The CrowdStrike AIDR shape is read from its policy source, not a live capture."
    )

    def __init__(
        self,
        message: str,
        *,
        categories: list[str] | None = None,
        provider: str | None = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.categories = categories or []
        self.provider = provider


class PIIDetected(PolicyViolation):
    """The PII-detection policy blocked the request or response.

    ``entities`` lists the flagged entity types (e.g. ``["Email"]``). The message
    (``str(exc)``, ``exc.args``, ``repr(exc)``) names those types with their count
    and character offsets but never the flagged values, so logging the exception
    or printing it in ``donkey doctor`` does not re-emit what the policy blocked.

    ``gateway_message`` is the gateway's own rejection text, which echoes every
    flagged value verbatim; it is kept for callers that need it and is not
    rendered by ``str()`` or ``repr()``. The raw body is also on ``.response``.
    Treat both as carrying the blocked content.

    Terminal and never retried by the SDK, like every :class:`PolicyViolation`.

    Docs: https://docs.donkey-kit.dev/errors
    """

    policy = "pii-detection"
    remediation: str = (
        "The PII-detection policy blocked this request because the prompt "
        "(or completion) contained personally identifiable information. "
        "Remove or redact the flagged values, or relax the policy's entity "
        "list / action in API Manager."
    )

    def __init__(
        self,
        message: str,
        *,
        entities: list[str] | None = None,
        gateway_message: str | None = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.entities = entities or []
        self.gateway_message = gateway_message


class AgentKilled(PolicyViolation):
    """The gateway's Agent Kill Switch blocked this agent — it is quarantined in
    Governance > Security, or listed in the policy's *Killed Agent IDs*.

    The upstream model was never called: the gateway refused the agent itself,
    not the request, so no change to the prompt or parameters can succeed. The
    live capture carries only ``code`` + ``message`` — no kill reason (#314)."""

    policy = "agent-kill-switch"
    remediation: str = (
        "An administrator has blocked this agent via the Agent Kill Switch. "
        "This is terminal and was NOT retried; ask them to restore model "
        "access in Governance > Security."
    )


class UpstreamModelError(DonkeyError):
    """Provider-side failure (5xx). Retryable."""

    remediation: str = (
        "The upstream provider failed (5xx); the transport already retried a "
        "503, and a 502/504 too unless the request was a model call (retried "
        "only with retry_model_calls_on_gateway_errors). Retry later, and if it "
        "persists escalate to the provider quoting .request_id."
    )


class UpstreamRequestError(DonkeyError):
    """The upstream provider rejected the request (4xx), passed through the
    gateway verbatim (e.g. OpenAI ``model_not_found``). This is a client-side
    mistake, NOT a gateway policy refusal and NOT a provider outage, so it is
    terminal (never retried) but distinct from :class:`PolicyViolation`.

    Carries the provider's own ``code``/``type``/``param`` when present so the
    caller can act (fix the model, the params, etc.).

    ``remediation`` is a class attribute so ``donkey doctor`` (#202) has one
    canonical wording for a model-rejected diagnosis (the
    ``model_not_found`` passthrough, docs/verified-apis.md §4) rather than a
    second copy."""

    #: Single source of next-step wording for a rejected-request diagnosis. The
    #: common case doctor keys on is ``code == "model_not_found"``.
    remediation: str = (
        "The upstream provider rejected the request (a 4xx passed through the "
        "gateway). If `code`/`param` names the model (e.g. model_not_found), the "
        "requested model is not available on this proxy — request it in API "
        "Manager or choose a model this instance routes. Otherwise fix the "
        "flagged parameter (inspect .code / .param / .response)."
    )

    def __init__(
        self,
        message: str,
        *,
        code: str | None = None,
        error_type: str | None = None,
        param: str | None = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.code = code
        self.error_type = error_type
        self.param = param


class ModelNotRoutable(DonkeyError):
    """The gateway could not pick a provider for the requested model, so it
    rejected the request before any upstream call (#825).

    Seen on a model-based-routing proxy with more than one provider configured:
    a bare model name (``gpt-5-mini``) is ambiguous unless the gateway knows it
    from exactly one provider, so it answers ``400`` with a flat-string
    ``error`` asking for ``provider/model`` (docs/verified-apis.md §4). This is a
    client configuration mistake, NOT a gateway policy refusal (so it is not a
    :class:`PolicyViolation`) and NOT an upstream rejection (the provider was
    never called, so it is not an :class:`UpstreamRequestError`). Terminal,
    never retried.

    ``model`` is the name the gateway echoed back, or ``None`` when it could not
    be read from the message."""

    #: Single source of next-step wording for the bare-model-name rejection.
    remediation: str = (
        "This proxy routes to more than one provider, so it cannot resolve a bare "
        "model name. Use the 'provider/model' form for the model, e.g. "
        "openai/gpt-5-mini instead of gpt-5-mini."
    )

    def __init__(self, message: str, *, model: str | None = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        self.model = model


class GatewayUnavailable(DonkeyError):
    """The gateway could not be reached at all — a transport-level failure (DNS,
    refused connection, TLS error, timeout) with NO HTTP response behind it
    (BG §1.2, #379).

    This is the one *ungoverned* failure the taxonomy names. Every other error
    here describes something the gateway told us; this one is the gateway not
    being there to tell us anything. Deliberately NOT a :class:`PolicyViolation`:
    nothing was refused — the request never reached a policy — so surfacing it as
    a refusal would misrepresent it at the framework boundary. Giving it a type
    lets a long-running agent distinguish "lost the gateway" from any other
    network fault and react — checkpoint, queue, shed load, or fall back to a
    non-AI path — instead of pattern-matching a raw ``httpx.TransportError``.

    Carries ``base_url`` (the origin that failed, so a handler can act on it
    without re-parsing the message) and ``cause`` (the underlying httpx
    exception, also chained via ``raise ... from``). ``request_id`` is inherited
    from :class:`DonkeyError` and is always ``None`` here — there was no response
    to read the gateway's own id from — while ``correlation_id`` / ``call_id``
    (the ids the client sent) are populated from the request that failed.

    ``remediation`` is a class attribute so it has a single canonical wording,
    shared with ``donkey doctor`` (#202) rather than duplicated."""

    #: The three real causes, named, plus the pointer to the startup/CI
    #: diagnostic. Single source of wording for this failure (#202, #379).
    remediation: str = (
        "The gateway could not be reached and no HTTP response came back. The "
        "three usual causes: (1) the host is unreachable — DNS failure or the "
        "gateway is down; (2) the configured base URL is wrong; or (3) network "
        "egress to the gateway is blocked — a firewall or air-gapped environment. "
        "Run `donkey doctor` to diagnose connectivity, and check `base_url` on "
        "this error against your gateway's address."
    )

    def __init__(
        self,
        message: str,
        *,
        base_url: str | None = None,
        cause: BaseException | None = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.base_url = base_url
        self.cause = cause


def gateway_unavailable(
    *,
    base_url: str | None = None,
    cause: BaseException | None = None,
    correlation_id: str | None = None,
    call_id: str | None = None,
) -> GatewayUnavailable:
    """Build a :class:`GatewayUnavailable` with the standard message from a
    transport-level failure. The message names the origin and the underlying
    cause; the full remediation lives on the returned exception's
    :attr:`GatewayUnavailable.remediation`."""
    where = f" at {base_url}" if base_url else ""
    detail = f": {cause}" if cause is not None and str(cause) else "."
    return GatewayUnavailable(
        f"The gateway could not be reached{where}{detail}",
        base_url=base_url,
        cause=cause,
        correlation_id=correlation_id,
        call_id=call_id,
    )


class ToolInvocationError(DonkeyError):
    """An MCP tool call failed."""

    remediation: str = (
        "An MCP tool call failed. Check the tool's arguments and the server's "
        "reported error in the message, then retry or handle the failure in the agent."
    )


class RegistryError(DonkeyError):
    """Exchange discovery / resolution failed."""

    remediation: str = (
        "Exchange discovery or resolution failed. Check the asset reference "
        "(group, asset id, version) and that the control-plane credentials can "
        "read it in Exchange."
    )


class PublicationDrift(DonkeyError):
    """verify(): the live server no longer matches the Exchange descriptor (BG §2.5)."""

    remediation: str = (
        "The live server no longer matches its Exchange descriptor. Republish the "
        "descriptor, or roll the server back, so the two agree."
    )


def classify(
    response: ResponseLike,
    *,
    correlation_id: str | None = None,
    call_id: str | None = None,
) -> DonkeyError:
    """Map an HTTP error response to a specific exception.

    ``correlation_id`` (the run id) and ``call_id`` (the per-request id) are the two ids the
    client sent (BG §1.1, #195). When not passed explicitly they are read back from the
    response's own request headers, so a caller bridging an ``openai.APIStatusError`` —
    ``classify(err.response)`` — gets a :class:`DonkeyError` whose ``correlation_id`` equals the
    header that was sent with no extra wiring. An explicit argument (e.g. when the header name
    was overridden via config) always wins over the auto-derived value.

    The precise policy discrimination (BG §1.2, working instruction #4) is driven by real
    rejection captures from a live governed LLM proxy (docs/verified-apis.md §4), or, where
    no proxy can produce one, by the gateway policy's own source (ADR 0016), NOT guesses:

    * **PII detection** rejects with **403** and a *nested* error object whose ``type`` is
      ``"pii_detected"`` (and, unlike a genuine auth failure, NO ``www-authenticate`` header).
      So a 403 is NOT automatically an auth error — the error ``type`` is checked first. The
      gateway's message echoes each flagged value, so the exception message is rebuilt from the
      entity types and offsets and the original is kept on ``PIIDetected.gateway_message``.
    * **Agent Kill Switch** rejects with **403** and a *nested* error object whose ``code`` is
      ``"agent_killed"`` (no ``type``, no ``www-authenticate``) → :class:`AgentKilled`. Keyed on
      the body ``code``, not the status, and checked BEFORE the auth and generic-4xx rules so a
      killed agent is not mis-typed as an :class:`UpstreamRequestError` (#694).
    * **Request rate limit** rejects with **429**, the body ``{"error":"Too Many Requests"}``
      and the ``x-ratelimit-limit`` / ``x-ratelimit-remaining`` / ``x-ratelimit-reset`` (ms)
      headers → :class:`RequestRateLimitExceeded` (#974). Keyed on those headers, which the
      policy sends only with ``exposeHeaders: true``; without them the 429 is not told apart.
    * **Token rate limit** rejects with **429** and an **empty body**; the budget state is
      entirely in headers (``x-token-limit`` / ``x-token-remaining`` / ``x-token-reset`` in ms).
      There is NO standard ``retry-after``. Any 429 the request-rate-limit rule does not claim
      maps here.
    * **client-id-enforcement** rejects with **401** + a *flat-string* ``error``
      and a ``www-authenticate`` header → auth.
    * **Upstream provider passthrough** (e.g. OpenAI ``model_not_found``) is a
      non-429 4xx with a nested error object carrying ``code``/``type``/``param``.
    * **Bare model name on a multi-provider proxy** rejects with **400** and a
      *flat-string* ``error`` saying the model "is not in the known unique model
      map" → :class:`ModelNotRoutable` (a client mistake, not a policy refusal;
      #825). Keyed on that gateway sentence.
    * **Injection protection** rejects with the ``x-injection-protection:
      blocked`` header (the header, not the status, is the discriminator; #181)
      → :class:`PromptInjectionBlocked`.
    * **Regex Prompt Guard** rejects with **403** and a *flat-string* ``error`` plus a top-level
      ``matched_patterns`` list → :class:`PromptInjectionBlocked` (``policy="regex-prompt-guard"``).
      Keyed on ``matched_patterns`` and checked BEFORE the 401/403→auth rule so a deny-list
      block is not mis-typed as auth.
    * **Content safety / guardrails** (Azure Content Safety, Amazon Bedrock Guardrails, Google
      Model Armor; CrowdStrike AIDR from its policy source) reject with **403** and a vendor
      header ``x-llm-proxy-<vendor>-...-action: reject`` → :class:`ContentSafetyBlocked`
      (flagged reasons parsed from the sibling ``...-reason`` header). Also checked before the
      auth rule; on a fail-closed 503 it is a 5xx outage.
    * **OAuth2 OBO CIBA step-up** answers **401** ``error: authorization_pending`` while the user
      has not approved, and **403** ``error: access_denied`` once they deny →
      ``GatewayApprovalPending`` / ``GatewayApprovalDenied`` (``core/step_up``, #1320). Also
      checked before the auth rule: the credential was accepted.

    These shapes have no ``_verify.py`` constant — ``classify()`` reads them straight from the
    response — so their verification record is the docs/verified-apis.md §4 rows. Any other
    content-moderation / federated-guardrail shape falls through to a generic
    :class:`PolicyViolation` whose message names what was observed and says the shape is
    unconfirmed (#184). Because auth is the verified 401 / ``www-authenticate`` shape, a **403
    carrying no ``www-authenticate`` header** and matching none of the policy discriminators
    above is treated as one of these unconfirmed refusals, not mis-typed as an :class:`AuthError`.

    A shape a customer registered with ``classify.register()`` (BG §2.2, #210,
    ``core/classifier``) is matched after every built-in above and before the upstream-envelope
    rule, so a verified shape always wins and a custom policy's nested-envelope 403 is not
    mis-typed as an :class:`UpstreamRequestError`.
    """

    shape = _shape(response, correlation_id=correlation_id, call_id=call_id)
    for matcher in _MATCHERS:
        error = matcher(shape)
        if error is not None:
            return error
    return DonkeyError(f"Unexpected response ({shape.status}).", **shape.kw)


@dataclass(frozen=True)
class _Shape:
    """What every :func:`classify` matcher reads off one response, parsed once.

    ``body`` is the top-level JSON object (it also carries the Regex Prompt Guard's
    ``matched_patterns`` key), ``None`` when the body is not a JSON object, e.g. Gemini's
    LIST-shaped error envelope (#548), whose nested error object ``nested_error`` still recovers.
    ``error_obj`` is the nested ``error`` object from either envelope shape, iff it is itself an
    object. ``error_type`` is its ``type`` (the OpenAI-format discriminator: ``pii_detected``,
    ``invalid_request_error`` …), or Gemini's string ``status`` (``INVALID_ARGUMENT``), which
    stands in for it (#548). ``kw`` are the constructor kwargs every classified error shares."""

    response: ResponseLike
    status: int
    body: dict[str, Any] | None
    error_obj: dict[str, Any] | None
    error_type: str | None
    kw: dict[str, Any]


def _shape(response: ResponseLike, *, correlation_id: str | None, call_id: str | None) -> _Shape:
    """Parse ``response`` once for :func:`classify`'s matchers."""
    sent_correlation, sent_call = sent_ids(response)
    error_obj = nested_error(response)
    return _Shape(
        response=response,
        status=response.status_code,
        body=json_body(response),
        error_obj=error_obj,
        error_type=(
            str_or_none(error_obj.get("type") or error_obj.get("status"))
            if error_obj is not None
            else None
        ),
        kw={
            # Explicit arg wins (e.g. a config-overridden header name); else the
            # id the client actually sent, read back from the request (BG §1.1, #195).
            "correlation_id": correlation_id if correlation_id is not None else sent_correlation,
            "call_id": call_id if call_id is not None else sent_call,
            # The upstream provider's request id — resolved from the per-provider header list, so
            # a Bedrock refusal (x-amzn-requestid, no x-request-id) is not silently None (#542).
            "request_id": read_request_id(response),
            "response": response,
        },
    )


# Next-step wording for the two classified shapes whose class default is not the right
# sentence: one constant each, never inline. Every other shape uses its class's own
# ``remediation`` (#728).
_REGEX_PROMPT_GUARD_REMEDIATION = (
    "The Regex Prompt Guard policy matched a denied pattern in the prompt. Remove "
    "or rephrase the flagged content, or adjust the policy's deny-list patterns in "
    "API Manager."
)
_UNCONFIRMED_REFUSAL_REMEDIATION = (
    "This refusal matched no documented rejection shape, so its contract is "
    "unconfirmed (#184). It is terminal and was NOT retried. Please file an "
    "issue on the donkey-development-kit repo with the response status, headers "
    "and body (all carried on this exception's .response) so the shape can be typed."
)


def _match_pii(s: _Shape) -> DonkeyError | None:
    """Gateway PII policy: 403 + nested object, ``type == "pii_detected"``. Ahead
    of the auth rule, because a PII block is not an auth failure. The gateway's
    message echoes each flagged value, so the message is rebuilt from the entity
    types and offsets."""
    if s.error_type != "pii_detected":
        return None
    gateway_message = str_or_none(s.error_obj.get("message")) if s.error_obj else None
    spans = pii_spans(gateway_message)
    return PIIDetected(
        pii_summary(s.status, spans),
        entities=[entity for entity, _start, _end in spans],
        gateway_message=gateway_message,
        **s.kw,
    )


def _match_agent_killed(s: _Shape) -> DonkeyError | None:
    """Agent Kill Switch: 403 + nested object, ``code == "agent_killed"`` (no
    ``type``; docs/verified-apis.md §4, #694). Keyed on the body code, ahead of
    the auth and generic-4xx rules: a killed agent is neither."""
    if s.error_obj is None or s.error_obj.get("code") != "agent_killed":
        return None
    message = str_or_none(s.error_obj.get("message"))
    return AgentKilled(message or f"Agent blocked by an active kill switch ({s.status}).", **s.kw)


def _match_content_safety(s: _Shape) -> DonkeyError | None:
    """Content safety / guardrails: 403 + a vendor ``...-action: reject`` header
    (docs/verified-apis.md §4, #253/#568/#1178/#1176/#1183/#1429). Keyed on the header, so an odd
    or absent body is still caught; ahead of the auth rule; never a fail-closed 503."""
    cs = content_safety_reject(s.response)
    if cs is None:
        return None
    provider, vendor, categories = cs
    cats = f" ({', '.join(categories)})" if categories else ""
    return ContentSafetyBlocked(
        f"Request blocked by {vendor}{cats} ({s.status}).",
        categories=categories,
        provider=provider,
        **s.kw,
    )


def _match_regex_prompt_guard(s: _Shape) -> DonkeyError | None:
    """Regex Prompt Guard: 403 + a top-level ``matched_patterns`` list (a
    flat-string ``error``, not the nested upstream envelope;
    docs/verified-apis.md §4, #253). Ahead of the auth rule."""
    matched = s.body.get("matched_patterns") if s.body is not None else None
    if not isinstance(matched, list):
        return None
    pats = ", ".join(str(p) for p in matched)
    return PromptInjectionBlocked(
        f"Request blocked by the regex prompt-guard policy ({s.status})"
        + (f": matched {pats}." if pats else "."),
        policy="regex-prompt-guard",
        remediation=_REGEX_PROMPT_GUARD_REMEDIATION,
        **s.kw,
    )


def _match_injection_protection(s: _Shape) -> DonkeyError | None:
    """Injection protection: the ``x-injection-protection: blocked`` header, not
    the status (#181, docs/verified-apis.md §4). Ahead of the generic 4xx rule
    so it wins even over an upstream-shaped body; a 400 without the header is
    never :class:`PromptInjectionBlocked`."""
    if s.response.headers.get(_wire.INJECTION_PROTECTION_HEADER) != "blocked":
        return None
    return PromptInjectionBlocked(
        f"Request blocked by the injection-protection policy ({s.status}).", **s.kw
    )


def _match_auth(s: _Shape) -> DonkeyError | None:
    """Auth: the verified client-id-enforcement shape (docs/verified-apis.md §4), a 401 or a 403
    carrying a ``www-authenticate`` challenge. A 403 without it falls through to the
    unconfirmed-refusal rule, not auth (#184); a CIBA 401 is claimed earlier (#1320)."""
    challenged = _wire.WWW_AUTHENTICATE_HEADER in s.response.headers
    if not (s.status == 401 or (s.status == 403 and challenged)):
        return None
    return AuthError(
        f"Authentication/authorization failed ({s.status}). Check the consumer "
        "client_id/client_secret pair and its API Manager authorization "
        "for this LLM-proxy instance (see docs/verified-apis.md §2).",
        **s.kw,
    )


def _match_request_rate_limit(s: _Shape) -> DonkeyError | None:
    """Request rate limit: 429 with the gateway's ``x-ratelimit-limit`` / ``x-ratelimit-remaining``
    headers and no ``x-token-*`` trio (docs/verified-apis.md §4, #974). Ahead of the token
    rule, which takes any other 429."""
    headers = s.response.headers
    if (
        s.status != 429
        or _wire.TOKEN_LIMIT_HEADER in headers
        or _wire.REQUEST_LIMIT_HEADER not in headers
        or _wire.REQUEST_REMAINING_HEADER not in headers
    ):
        return None
    return RequestRateLimitExceeded(
        "Request rate limit exceeded (429).",
        retry_after=read_retry_after(s.response, reset_header=_wire.REQUEST_RESET_HEADER),
        **s.kw,
    )


def _match_token_budget(s: _Shape) -> DonkeyError | None:
    """Token rate limit: 429 with an empty body; the budget state is header-only
    (``x-token-limit`` / ``x-token-remaining`` / ``x-token-reset`` ms), with no
    ``retry-after``."""
    if s.status != 429:
        return None
    return TokenBudgetExceeded(
        "Token rate limit or budget exceeded (429).",
        retry_after=read_retry_after(s.response),
        **s.kw,
    )


def _match_upstream_request(s: _Shape) -> DonkeyError | None:
    """Any other 4xx with the upstream provider's nested envelope: a request
    mistake passed through the gateway (e.g. ``model_not_found``)."""
    if not 400 <= s.status < 500 or s.error_obj is None:
        return None
    return UpstreamRequestError(
        f"The upstream model provider rejected the request ({s.status}): "
        f"{s.error_obj.get('message') or 'see .response'}",
        # OpenAI sends a string ``code`` (``model_not_found``); Gemini a numeric
        # one (400). Carry both, stringified (#548).
        code=code_str(s.error_obj.get("code")),
        error_type=s.error_type,
        param=str_or_none(s.error_obj.get("param")),
        **s.kw,
    )


def _match_model_not_routable(s: _Shape) -> DonkeyError | None:
    """A bare model name on a multi-provider proxy: a 4xx flat-string ``error`` naming the model
    and asking for ``provider/model`` (docs/verified-apis.md §4, #825). Keyed on the gateway's
    own sentence, since the status and envelope match the unconfirmed fall-through."""
    if not 400 <= s.status < 500:
        return None
    flat_error = str_or_none(s.body.get("error")) if s.body is not None else None
    unroutable = MODEL_NOT_ROUTABLE_RE.search(flat_error) if flat_error else None
    if unroutable is None:
        return None
    return ModelNotRoutable(
        f"The gateway could not route the requested model ({s.status}): {flat_error}",
        model=unroutable.group("model"),
        **s.kw,
    )


def _match_unconfirmed_refusal(s: _Shape) -> DonkeyError | None:
    """An unrecognised non-429 4xx with no nested provider envelope: a refusal whose contract
    cannot be pinned (#253). Surfaced honestly, naming the status and any gateway policy
    headers, never coerced into an unverified subclass (#184)."""
    if not 400 <= s.status < 500:
        return None
    policy_headers = sorted(
        name
        for name in s.response.headers
        if name.lower().startswith(_wire.LLM_PROXY_HEADER_PREFIX)
    )
    observed = f"status {s.status}"
    if policy_headers:
        observed += f"; policy headers: {', '.join(policy_headers)}"
    return PolicyViolation(
        f"Request refused by a gateway policy; shape unconfirmed ({observed}). "
        "It matched no documented rejection contract.",
        policy="unknown",
        remediation=_UNCONFIRMED_REFUSAL_REMEDIATION,
        **s.kw,
    )


def _match_upstream_failure(s: _Shape) -> DonkeyError | None:
    """A 5xx: a provider-side failure."""
    if not 500 <= s.status < 600:
        return None
    return UpstreamModelError(f"Upstream/provider failure ({s.status}). Retryable.", **s.kw)


#: A customer-registered refusal shape (BG §2.2, #210), installed by ``core/classifier`` when
#: it is imported. A hook slot, not an import: this module sits below the registry in the core
#: layers, and with no registry imported there is nothing registered to match.
_REGISTERED_MATCHER: list[Callable[[_Shape], DonkeyError | None]] = []

#: The OBO policy's CIBA pending/denied rules (#1320), installed by ``core/step_up``, whose
#: types subclass this module's. ``core`` imports it, so the slot is always filled.
_STEP_UP_MATCHERS: list[Callable[[_Shape], DonkeyError | None]] = []


def _match_step_up(s: _Shape) -> DonkeyError | None:
    """A CIBA step-up answer (#1320): a 401/403 keyed on its flat-string body ``error``."""
    return next((e for m in _STEP_UP_MATCHERS if (e := m(s)) is not None), None)


def _match_registered(s: _Shape) -> DonkeyError | None:
    """A shape a customer registered with ``classify.register()`` (#210)."""
    return _REGISTERED_MATCHER[0](s) if _REGISTERED_MATCHER else None


#: The built-in rules a registration may never shadow (``classify.register()`` rejects any spec
#: one of these claims). Every policy shape with its own discriminator (body ``type``/``code``/
#: ``error``, a vendor header, ``matched_patterns``) precedes the 401/403 auth rule.
#: ``model_not_routable`` needs a flat-string ``error`` and ``upstream_request`` a nested
#: object, so the two are disjoint and their order between them is free.
_POLICY_MATCHERS: tuple[Callable[[_Shape], DonkeyError | None], ...] = (
    _match_pii,
    _match_agent_killed,
    _match_content_safety,
    _match_regex_prompt_guard,
    _match_injection_protection,
    _match_step_up,
    _match_auth,
    _match_request_rate_limit,
    _match_token_budget,
    _match_model_not_routable,
)

#: :func:`classify`'s rules, most specific first. The order is the contract: the built-ins,
#: then the customer registry (so a verified shape always wins, and a registered nested-envelope
#: refusal is not mis-typed as a provider mistake), then the upstream envelope and the
#: unconfirmed-refusal fall-through.
_MATCHERS: tuple[Callable[[_Shape], DonkeyError | None], ...] = (
    *_POLICY_MATCHERS,
    _match_registered,
    _match_upstream_request,
    _match_unconfirmed_refusal,
    _match_upstream_failure,
)