tracer-cloud--opensre/core/llm/failure_classification.py

"""Shared LLM failure string classification (provider-agnostic)."""

from __future__ import annotations

import re

# Patterns ordered by specificity — first match wins in classify_cli_failure_category_hint.
_QUOTA_RE = re.compile(
    r"quota|rate.?limit|429|too many request|insufficient_quota|"
    r"out of credit|billing|usage limit|spending limit|plan limit|"
    r"exceeded.*limit|limit.*exceeded|maximum.*usage",
    re.IGNORECASE,
)
_AUTH_RE = re.compile(
    r"unauthorized|401|invalid.?api.?key|api.?key.*invalid|"
    r"authentication.?fail|not authenticated|not logged.?in|"
    r"no credentials|token.*expired|expired.*token|invalid.?token|"
    r"permission denied|access denied|403|forbidden",
    re.IGNORECASE,
)
_CONTEXT_OVERFLOW_RE = re.compile(
    r"context.?length|context.?window|max(?:imum)?\s+context\s+length|"
    r"max.?token|token.?limit|prompt.*too\s+long|prompt.*too.?large|"
    r"input.*exceed|reduce.*context|string too long",
    re.IGNORECASE,
)
_NETWORK_RE = re.compile(
    r"network.*error|connection.*refused|dns.*fail|unreachable|"
    r"no route to host|connection reset|ssl.*error|certificate.*error|"
    r"name.*resolution|getaddrinfo",
    re.IGNORECASE,
)
_ERROR_KEYWORD_RE = re.compile(r"error|fail|exception|invalid", re.IGNORECASE)

_SILENT_FAILURE_HINT = (
    "no error detail from the CLI — most likely quota exhausted or expired auth; "
    "check your plan/credits or re-login"
)


def is_context_length_overflow(message: str) -> bool:
    """Return True when *message* indicates prompt/context/token limit exhaustion.

    Avoids bare ``too long`` so timeout strings like "request took too long"
    are not misclassified as context overflow.
    """
    return _CONTEXT_OVERFLOW_RE.search(message) is not None


def classify_cli_failure_category_hint(stdout: str, stderr: str, _returncode: int) -> str | None:
    """Return a category hint (quota/auth/context/network) when output matches."""
    combined = f"{stdout}\n{stderr}".strip()

    if _QUOTA_RE.search(combined):
        return "quota or rate limit exceeded — check your plan/billing or wait before retrying"
    if _AUTH_RE.search(combined):
        return "authentication failed — verify your API key or re-login with the provider CLI"
    if is_context_length_overflow(combined):
        return (
            "prompt too long — shorten the input or reduce accumulated context "
            "(/context to inspect)"
        )
    if _NETWORK_RE.search(combined):
        return "network error — check connectivity and provider status"
    return None


def classify_cli_failure_hint(stdout: str, stderr: str, returncode: int) -> str | None:
    """Return a short actionable hint for a known failure category, or None."""
    category = classify_cli_failure_category_hint(stdout, stderr, returncode)
    if category is not None:
        return category

    combined = f"{stdout}\n{stderr}".strip()
    if returncode not in (0, 130) and (
        not combined or (len(combined) < 120 and not _ERROR_KEYWORD_RE.search(combined))
    ):
        return _SILENT_FAILURE_HINT

    return None


__all__ = [
    "classify_cli_failure_category_hint",
    "classify_cli_failure_hint",
    "is_context_length_overflow",
]