src/free_claude_code/core/diagnostics.py

"""Credential-safe diagnostics shared across product boundaries."""
from __future__ import annotations


import json
import re
import traceback
from dataclasses import dataclass
from typing import Any

from .failures import ExecutionFailure

ERROR_DETAIL_DISPLAY_CAP_BYTES = 16_384
_MAX_CAUSE_CHAIN_DEPTH = 4
_UPSTREAM_BODY_ATTR = "_fcc_upstream_error_body"
_UPSTREAM_BODY_TRUNCATED_ATTR = "_fcc_upstream_error_body_truncated"

_SECRET_TEXT_REPLACEMENTS = (
    (
        re.compile(
            r"(?i)(?P<prefix>[\"']?authorization[\"']?\s*[:=]\s*)"
            r"(?P<quote>[\"']?)(?:(?:bearer|basic)\s+)?"
            r"[^\"'\s,;&}\]]+(?P=quote)"
        ),
        r"\g<prefix>\g<quote><redacted>\g<quote>",
    ),
    (
        re.compile(
            r"(?i)(?P<prefix>[\"']?(?:api[_-]?key|access[_-]?token|"
            r"refresh[_-]?token|token|client[_-]?secret|secret|password)"
            r"[\"']?\s*[:=]\s*)(?P<quote>[\"']?)"
            r"[^\"'\s,;&}\]]+(?P=quote)"
        ),
        r"\g<prefix>\g<quote><redacted>\g<quote>",
    ),
    (re.compile(r"(?i)(bearer\s+)[^\s,;]+"), r"\1<redacted>"),
    (
        re.compile(
            r"(?i)(?<![a-z0-9])(?:sk-[a-z0-9._-]{8,}|"
            r"nvapi-[a-z0-9._-]{8,}|hf_[a-z0-9_-]{8,}|"
            r"gsk_[a-z0-9_-]{8,}|github_pat_[a-z0-9_]{8,}|"
            r"gh[pousr]_[a-z0-9]{8,}|AIza[a-z0-9_-]{20,})"
            r"(?![a-z0-9])"
        ),
        "<redacted>",
    ),
)


@dataclass(frozen=True, slots=True)
class UpstreamErrorDetail:
    """Sanitized diagnostic detail extracted from an upstream exception."""


    status_code: int | None = None
    body_text: str | None = None
    exception_text: str | None = None
    cause_chain_text: str | None = None
    category_hint: str | None = None
    body_truncated: bool = False


def redact_sensitive_error_text(text: str) -> str:
    """Redact recognizable credentials while preserving diagnostic context."""
    sanitized = text
    for pattern, replacement in _SECRET_TEXT_REPLACEMENTS:
        sanitized = pattern.sub(replacement, sanitized)
    return sanitized


def safe_exception_message(
    exc: BaseException,
    *,
    fallback: str = "Provider request failed unexpectedly.",
) -> str:
    """Return a redacted, non-empty exception message."""
    message = redact_sensitive_error_text(str(exc).strip())
    return message or fallback


def format_user_error_preview(exc: BaseException, *, max_len: int = 200) -> str:
    """Return a short redacted exception preview for chat surfaces."""
    return safe_exception_message(exc)[:max_len]


def attach_upstream_error_body(
    exc: Exception,
    body: bytes | str,
    *,
    truncated: bool = False,
) -> None:
    """Attach a bounded streamed response body for later safe formatting."""
    setattr(exc, _UPSTREAM_BODY_ATTR, body)
    setattr(exc, _UPSTREAM_BODY_TRUNCATED_ATTR, truncated)


def exception_cause_types(exc: BaseException) -> tuple[str, ...]:
    """Return exception cause type names without logging their contents."""
    return tuple(type(cause).__name__ for cause in _exception_causes(exc))


def redacted_exception_traceback(exc: BaseException) -> str:
    """Format a traceback while redacting recognizable credentials."""
    return redact_sensitive_error_text("".join(traceback.format_exception(exc)))


def extract_upstream_error_detail(exc: Exception) -> UpstreamErrorDetail:
    """Extract bounded, redacted body, exception, and cause-chain details."""
    raw_body = getattr(exc, _UPSTREAM_BODY_ATTR, None)
    body_truncated = bool(getattr(exc, _UPSTREAM_BODY_TRUNCATED_ATTR, False))
    if raw_body is None:
        raw_body = getattr(exc, "body", None)
    if raw_body is None:
        raw_body = _body_from_response(exc)

    body_text = _normalize_body_text(raw_body)
    if body_text is not None:
        body_text = redact_sensitive_error_text(body_text)
        body_text, capped = _cap_text_bytes(body_text)
        body_truncated = body_truncated or capped

    exception_text = str(exc).strip() or None
    if exception_text is not None:
        exception_text = redact_sensitive_error_text(exception_text)
        exception_text, _ = _cap_text_bytes(exception_text)

    return UpstreamErrorDetail(
        status_code=_status_code_from_exception(exc),
        body_text=body_text,
        exception_text=exception_text,
        cause_chain_text=_exception_cause_chain_text(exc),
        category_hint=_category_hint_from_body(raw_body, body_text),
        body_truncated=body_truncated,
    )


def format_execution_failure_message(
    failure: ExecutionFailure,
    detail: UpstreamErrorDetail,
    *,
    upstream_name: str,
    request_id: str | None = None,
) -> str:
    """Build a copyable, redacted diagnostic for a finalized execution failure."""
    stable_message = failure.message
    has_upstream_detail = detail.status_code is not None or detail.body_text is not None
    if not has_upstream_detail:
        lines = [stable_message]
        if detail.exception_text and detail.exception_text != stable_message:
            lines.extend(("", "Provider exception:", detail.exception_text))
        if detail.cause_chain_text:
            lines.extend(("", "Caused by:", detail.cause_chain_text))
        _append_request_id_lines(lines, request_id)
        return "\n".join(lines)

    if detail.status_code == 405:
        lines = [
            f"Upstream provider {upstream_name} rejected the request method "
            "or endpoint (HTTP 405)."
        ]
    elif detail.status_code is not None:
        lines = [
            f"Upstream provider {upstream_name} returned HTTP {detail.status_code}."
        ]
    else:
        lines = [f"Upstream provider {upstream_name} returned an error."]

    lines.append(f"Category: {detail.category_hint or failure.kind.value}")
    if stable_message and stable_message != lines[0]:
        lines.append(f"Mapped message: {stable_message}")
    lines.extend(("", "Upstream error:"))
    lines.append(detail.body_text or "(empty upstream error body)")
    if _body_truncation_line_needed(detail):
        lines.append(f"... [truncated after {ERROR_DETAIL_DISPLAY_CAP_BYTES} bytes]")
    _append_request_id_lines(lines, request_id)
    return "\n".join(lines)


def _body_truncation_line_needed(detail: UpstreamErrorDetail) -> bool:
    """Return whether a separate truncation marker is needed."""
    return detail.body_truncated and (
        detail.body_text is None
        or f"truncated after {ERROR_DETAIL_DISPLAY_CAP_BYTES} bytes"
        not in detail.body_text
    )


def _status_code_from_exception(exc: Exception) -> int | None:
    status = getattr(exc, "status_code", None)
    if isinstance(status, int):
        return status
    response = getattr(exc, "response", None)
    response_status = getattr(response, "status_code", None)
    return response_status if isinstance(response_status, int) else None


def _body_from_response(exc: Exception) -> Any:
    response = getattr(exc, "response", None)
    if response is None:
        return None
    try:
        return response.json()
    except Exception:
        pass
    try:
        return response.text
    except Exception:
        return None


def _normalize_body_text(body: Any) -> str | None:
    if body is None:
        return None
    if isinstance(body, bytes):
        text = body.decode("utf-8", errors="replace")
    elif isinstance(body, str):
        text = body
    else:
        try:
            return json.dumps(body, ensure_ascii=False, separators=(",", ":"))
        except TypeError:
            text = str(body)
    stripped = text.strip()
    if not stripped:
        return None
    try:
        parsed = json.loads(stripped)
    except ValueError:
        return stripped
    return json.dumps(parsed, ensure_ascii=False, separators=(",", ":"))


def _cap_text_bytes(text: str) -> tuple[str, bool]:
    encoded = text.encode("utf-8", errors="replace")
    if len(encoded) <= ERROR_DETAIL_DISPLAY_CAP_BYTES:
        return text, False
    capped = encoded[:ERROR_DETAIL_DISPLAY_CAP_BYTES].decode("utf-8", errors="replace")
    return (
        f"{capped}\n... [truncated after {ERROR_DETAIL_DISPLAY_CAP_BYTES} bytes]",
        True,
    )


def _exception_causes(exc: BaseException) -> tuple[BaseException, ...]:
    causes: list[BaseException] = []
    seen = {id(exc)}
    current: BaseException | None = exc
    while current is not None and len(causes) < _MAX_CAUSE_CHAIN_DEPTH:
        next_exc = current.__cause__ or current.__context__
        if next_exc is None or id(next_exc) in seen:
            break
        seen.add(id(next_exc))
        causes.append(next_exc)
        current = next_exc
    return tuple(causes)


def _exception_cause_chain_text(exc: BaseException) -> str | None:
    lines: list[str] = []
    for cause in _exception_causes(exc):
        raw_text = str(cause).strip()
        lines.append(
            f"{type(cause).__name__}: {redact_sensitive_error_text(raw_text)}"
            if raw_text
            else type(cause).__name__
        )
    if not lines:
        return None
    text, _ = _cap_text_bytes("\n".join(lines))
    return text


def _category_hint_from_body(body: Any, body_text: str | None) -> str | None:
    parsed = body
    if isinstance(parsed, bytes):
        parsed = parsed.decode("utf-8", errors="replace")
    if isinstance(parsed, str):
        try:
            parsed = json.loads(parsed)
        except ValueError:
            parsed = None
    if isinstance(parsed, dict):
        error = parsed.get("error")
        if isinstance(error, dict):
            for key in ("type", "code"):
                value = error.get(key)
                if isinstance(value, str) and value.strip():
                    return value.strip()
        for key in ("type", "code"):
            value = parsed.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
    if (
        body_text
        and "model" in body_text.lower()
        and "unsupported" in body_text.lower()
    ):
        return "upstream_model_error"
    return None


def _append_request_id_lines(lines: list[str], request_id: str | None) -> None:
    if request_id:
        lines.extend(("", f"Request ID: {request_id}"))