litellm/litellm/_logging.py

import ast
import contextvars
import functools
import inspect
import itertools
import logging
import os
import re
import sys
from collections.abc import Iterator, Mapping
from datetime import datetime
from logging import Formatter
from typing import Final, TextIO
from urllib.parse import unquote

from pydantic import TypeAdapter

import litellm
from litellm.constants import (
    LITELLM_TRUNCATED_PAYLOAD_FIELD,
    LITELLM_TRUNCATION_STDOUT_SAFEGUARD_NOTE,
    MAX_BASE64_LENGTH_STDOUT_LOG,
    MAX_STRING_LENGTH_STDOUT_LOG,
)
from litellm.litellm_core_utils.env_utils import get_env_int
from litellm.litellm_core_utils.safe_json_dumps import UNSERIALIZABLE_OBJECT, safe_dumps, safe_json_structure
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
from litellm.litellm_core_utils.secret_redaction import (
    python_redact_string,
    python_redact_structured_value,
    redact_internal_details,
)
from litellm.litellm_core_utils.secret_redaction import (
    redact_string as redact_secret_string,
)
from litellm.rust_bridge import diagnostics

set_verbose = False

session_id_var: Final[contextvars.ContextVar[str]] = contextvars.ContextVar("session_id", default="")
trace_id_var: Final[contextvars.ContextVar[str]] = contextvars.ContextVar("trace_id", default="")

_MAX_CORRELATION_ID_LENGTH: Final = 256


def _sanitize_correlation_id(value: str) -> str:
    """Strip control characters, bound length, and redact credential-shaped
    content before a caller-controlled trace_id/session_id (e.g.
    litellm_session_id, x-litellm-trace-id) is stamped into log lines.

    Without the first two, a caller could embed \\r/\\n or terminal escape
    sequences to forge fake log entries, or submit an oversized value repeated
    across every log line for the request. Without the redaction, a caller
    could smuggle a real credential (e.g. an sk-... key) through this field:
    CorrelationContextFilter stamps trace_id/session_id onto the record after
    SecretRedactionFilter has already run, so those two fields never otherwise
    pass through credential redaction.
    """
    stripped: Final = "".join(ch for ch in value if ch.isprintable())
    return redact_string(stripped)[:_MAX_CORRELATION_ID_LENGTH]


def set_session_id(session_id: str) -> "contextvars.Token[str]":
    return session_id_var.set(_sanitize_correlation_id(session_id))


def set_trace_id(trace_id: str) -> "contextvars.Token[str]":
    return trace_id_var.set(_sanitize_correlation_id(trace_id))


if set_verbose is True:
    logging.warning(
        "`litellm.set_verbose` is deprecated. Please set `os.environ['LITELLM_LOG'] = 'DEBUG'` for debug logs."
    )

_ENABLE_SECRET_REDACTION: Final = os.getenv("LITELLM_DISABLE_REDACT_SECRETS", "").lower() != "true"


def redact_string(value: str) -> str:
    if not _ENABLE_SECRET_REDACTION:
        return value
    return redact_secret_string(value)


_redact_string = redact_string


_REDACTED_RECORD_ATTR: Final = "litellm_redacted"
_REDACTED_STAMP: Final = object()
_DECODED_LOG_PAYLOAD: Final = TypeAdapter(object)
_UNREDACTED_SCALAR_TYPES: Final = (bool, int, float, type(None))


def _is_redacted(record: logging.LogRecord) -> bool:
    return getattr(record, _REDACTED_RECORD_ATTR, None) is _REDACTED_STAMP


def _record_attributes(record: logging.LogRecord) -> Mapping[str, object]:
    return record.__dict__


def _scrubbing_changed_nothing(scrubbed: object, original: object) -> bool:
    try:
        return bool(scrubbed == original)
    except Exception:
        return False


def _plain_text(value: object) -> str:
    try:
        return str(value)
    except Exception:
        return UNSERIALIZABLE_OBJECT


def redact_secrets(value: str) -> str:
    """Public API: redact known secret/credential patterns from an arbitrary string.

    Use this for code paths that bypass the logging system — e.g. Slack/Teams
    alerting, HTTP error response bodies, or any other string that may contain
    secrets and will be sent to an external sink.

    Not to be confused with redact_message_input_output_from_logging() in
    litellm_core_utils/redact_messages.py, which redacts LLM prompt/response
    content for privacy — this function redacts credential patterns (API keys,
    PEM blocks, tokens, etc.) by shape.
    """
    if not _ENABLE_SECRET_REDACTION:
        return value
    return redact_string(value)


def redact_internal_details_from_client_message(value: str) -> str:
    """Public API: redact_secrets() plus filesystem paths, internal hostnames, and an
    embedded traceback, for a string about to leave the process in an HTTP response."""
    if not _ENABLE_SECRET_REDACTION:
        return value
    return redact_internal_details(value)


def _substituted_color_message(record: logging.LogRecord) -> str | None:
    """Render a record's ``color_message`` against its args, or None if absent.

    uvicorn's colorized formatter re-renders `color_message` against
    record.args at emit time (see uvicorn.logging.ColourizedFormatter) instead
    of using the already-formatted record.msg, so it has to be substituted
    before args are cleared or it is later formatted with no args and prints
    the raw "%s://%s:%d" placeholders instead of the URL.
    """
    color_message: Final = record.__dict__.get("color_message")
    if not isinstance(color_message, str) or not record.args:
        return None
    try:
        return color_message % record.args
    except Exception:
        return color_message


class SecretRedactionFilter(logging.Filter):
    """Scrubs known secret/credential patterns from log records."""

    _formatter = logging.Formatter()

    def filter(self, record: logging.LogRecord) -> bool:
        if not _ENABLE_SECRET_REDACTION or _is_redacted(record):
            return True
        return _process_record(record, base64_limit=0, text_limit=0, redact=True)


_secret_filter: Final = SecretRedactionFilter()


_MAX_SCRUBBED_ACCESS_ARG: Final = 512

_REDACTION_PLACEHOLDER: Final = "REDACTED"


def _hides_a_credential(value: str) -> bool:
    """Whether *value* only looks clean until it is percent-decoded."""
    decoded: Final = unquote(value)
    return python_redact_string(decoded) != decoded


def _drop_encoded_credential(scrubbed: str) -> str:
    """Drop the part of a request target that only decoding shows to be a secret.

    The request parser decodes query names and values, so `?k%65y=sk%2D...` is a
    working credential that the patterns, which match literal text, do not see.
    The decoded text is never logged back: it can carry a newline, and forging
    log lines is not a trade worth making for a readable request target.
    """
    path, separator, _query = scrubbed.partition("?")
    if _hides_a_credential(path):
        return _REDACTION_PLACEHOLDER
    if separator and _hides_a_credential(scrubbed):
        return f"{path}?{_REDACTION_PLACEHOLDER}"
    return scrubbed


def _scrub_access_arg(value: str) -> str:
    """Redact one access-log positional arg, bounding the scanned length.

    The request target is the only input to the secret regex an unauthenticated
    caller controls end to end, so it is cut back to a whole query parameter
    before it is scanned; a half-parameter would be too short to match its
    pattern and would then be logged raw.
    """
    if len(value) <= _MAX_SCRUBBED_ACCESS_ARG:
        return _drop_encoded_credential(python_redact_string(value))
    head: Final = value[:_MAX_SCRUBBED_ACCESS_ARG]
    kept: Final = head[: max(head.rfind("?"), head.rfind("&"))] if "?" in head else head
    scrubbed: Final = _drop_encoded_credential(python_redact_string(kept))
    return f"{scrubbed}... ({len(value) - len(kept)} more chars truncated) ..."


class AccessLogRedactionFilter(logging.Filter):
    """Scrubs known secret/credential patterns from HTTP access-log records.

    uvicorn's AccessFormatter unpacks ``record.args`` as a five-element tuple at
    emit time, so SecretRedactionFilter cannot be reused here: it collapses the
    record into ``record.msg`` and clears the args, and the formatter then raises.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if not _ENABLE_SECRET_REDACTION:
            return True
        if isinstance(record.args, tuple) and record.args:
            strings: Final = tuple(arg for arg in record.args if isinstance(arg, str))
            candidate: Final = diagnostics.run(
                lambda native: native.scrub_access_arguments(strings),
                lambda: tuple(_scrub_access_arg(arg) for arg in strings),
            )
            scrubbed: Final = (
                candidate if len(candidate) == len(strings) else tuple(_scrub_access_arg(arg) for arg in strings)
            )
            values: Final = iter(scrubbed)
            record.args = tuple(  # rebind-ok: a Filter scrubs records in place
                next(values) if isinstance(arg, str) else arg for arg in record.args
            )
            return True
        # No positional args means everything is in msg, where collapsing is correct.
        return _secret_filter.filter(record)


_access_log_filter: Final = AccessLogRedactionFilter()


@functools.lru_cache(maxsize=1)
def _parse_disabled_access_log_paths(raw: str) -> frozenset[str]:
    return frozenset(stripped for path in raw.split(",") if (stripped := path.strip()))


def _disabled_access_log_paths() -> frozenset[str]:
    """Read the variable per record so a value loaded later via proxy config
    environment_variables or dotenv is honored."""
    return _parse_disabled_access_log_paths(os.getenv("LITELLM_DISABLE_ACCESS_LOG_PATHS", ""))


class AccessLogPathFilter(logging.Filter):
    """Drops uvicorn.access records for request paths listed in LITELLM_DISABLE_ACCESS_LOG_PATHS.

    uvicorn passes record.args as (client_addr, method, full_path, http_version, status_code).
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if not isinstance(record.args, tuple) or len(record.args) < 3:
            return True
        full_path: Final = record.args[2]
        if not isinstance(full_path, str):
            return True
        return full_path.partition("?")[0] not in _disabled_access_log_paths()


_access_log_path_filter: Final = AccessLogPathFilter()


def _get_max_string_length_stdout_log() -> int:
    """Read the limit per record so a value loaded later via proxy config
    environment_variables is honored."""
    return get_env_int("MAX_STRING_LENGTH_STDOUT_LOG", MAX_STRING_LENGTH_STDOUT_LOG)


def _stdout_truncation_marker(skipped_chars: int) -> str:
    return (
        f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. "
        f"{LITELLM_TRUNCATION_STDOUT_SAFEGUARD_NOTE}) ..."
    )


def _truncate_for_stdout_log(text: str, limit: int) -> str:
    kept_chars: Final = limit - len(_stdout_truncation_marker(len(text)))
    if kept_chars <= 0:
        return text[:limit]
    head_chars: Final = kept_chars // 2
    tail_chars: Final = kept_chars - head_chars
    return f"{text[:head_chars]}{_stdout_truncation_marker(len(text) - kept_chars)}{text[-tail_chars:]}"


_BYTES_PER_KIB: Final = 1024
_BYTES_PER_MIB: Final = 1024 * 1024


def format_base64_size(num_chars: int) -> str:
    """Return a human-readable byte-size estimate from a base64 character count."""
    num_bytes: Final = num_chars * 3 / 4
    if num_bytes >= _BYTES_PER_MIB:
        return f"{num_bytes / _BYTES_PER_MIB:.2f}MB"
    if num_bytes >= _BYTES_PER_KIB:
        return f"{num_bytes / _BYTES_PER_KIB:.1f}KB"
    return f"{int(num_bytes)}B"


def _get_max_base64_length_stdout_log() -> int:
    return get_env_int("MAX_BASE64_LENGTH_STDOUT_LOG", MAX_BASE64_LENGTH_STDOUT_LOG)


@functools.lru_cache(maxsize=8)
def _base64_run_pattern(min_chars: int) -> "re.Pattern[str]":
    return re.compile(rf"(?<![A-Za-z0-9+/])[A-Za-z0-9+/]{{{min_chars},}}={{0,2}}")


_LOWER_HEX_DIGITS: Final = "0123456789abcdef"
_UPPER_HEX_DIGITS: Final = "0123456789ABCDEF"


def _looks_like_base64(run: str) -> bool:
    unpadded: Final = run.rstrip("=")
    is_hex_or_decimal: Final = not unpadded.strip(_LOWER_HEX_DIGITS) or not unpadded.strip(_UPPER_HEX_DIGITS)
    is_one_repeated_char: Final = not unpadded.strip(unpadded[0])
    return not is_hex_or_decimal or is_one_repeated_char


def _replace_base64_run(match: "re.Match[str]") -> str:
    run: Final = match.group(0)
    if not _looks_like_base64(run):
        return run
    return f"[base64_data truncated: {format_base64_size(len(run))}]"


def _collapse_base64_runs(text: str, limit: int) -> str:
    return _base64_run_pattern(limit + 1).sub(_replace_base64_run, text)


def _extra_structure(key: str, value: object) -> object:
    if isinstance(value, str):
        return value
    try:
        return safe_json_structure(value, key=key)
    except Exception:
        return _plain_text(value)


def _string_leaves(key: str | None, value: object) -> Iterator[tuple[str | None, str]]:
    if isinstance(value, str):
        yield key, value
    elif isinstance(value, dict):
        yield from itertools.chain.from_iterable(
            _string_leaves(child_key, child) for child_key, child in value.items() if isinstance(child_key, str)
        )
    elif isinstance(value, (list, tuple)):
        yield from itertools.chain.from_iterable(_string_leaves(key, child) for child in value)


def _replace_string_leaves(value: object, values: Iterator[str]) -> object:
    if isinstance(value, str):
        return next(values)
    if isinstance(value, dict):
        return {key: _replace_string_leaves(child, values) for key, child in value.items()}
    if isinstance(value, list):
        return [_replace_string_leaves(child, values) for child in value]
    if isinstance(value, tuple):
        return tuple(_replace_string_leaves(child, values) for child in value)
    return value


def _sort_processed_sets(original: object, processed: object) -> object:
    if isinstance(original, set) and isinstance(processed, list):
        return sorted(processed)
    if isinstance(original, dict) and isinstance(processed, dict):
        return {key: _sort_processed_sets(original.get(key), value) for key, value in processed.items()}
    if isinstance(original, list) and isinstance(processed, list):
        return [_sort_processed_sets(before, after) for before, after in zip(original, processed)]
    if isinstance(original, tuple) and isinstance(processed, tuple):
        return tuple(_sort_processed_sets(before, after) for before, after in zip(original, processed))
    return processed


def _python_process_diagnostic(
    message: str,
    exception: str | None,
    stack: str | None,
    leaves: tuple[tuple[str | None, str], ...],
    redact: bool,
    base64_limit: int,
    text_limit: int,
) -> tuple[str, str | None, str | None, tuple[str, ...], bool]:
    def process_text(text: str) -> str:
        collapsed: Final = _collapse_base64_runs(text, base64_limit) if base64_limit > 0 else text
        scrubbed: Final = python_redact_string(collapsed) if redact else collapsed
        return _truncate_for_stdout_log(scrubbed, text_limit) if 0 < text_limit < len(scrubbed) else scrubbed

    processed_message: Final = process_text(message)
    processed_exception: Final = process_text(exception) if exception is not None else None
    processed_stack: Final = python_redact_string(stack) if redact and stack is not None else stack
    processed_leaves: Final = tuple(
        python_redact_structured_value(key, text) if redact else text for key, text in leaves
    )
    changed: Final = (
        processed_message != message
        or processed_exception != exception
        or processed_stack != stack
        or any(processed != original for processed, (_, original) in zip(processed_leaves, leaves))
    )
    return processed_message, processed_exception, processed_stack, processed_leaves, changed


def _render_message(record: logging.LogRecord) -> str:
    try:
        return record.getMessage()
    except Exception:
        return record.msg if isinstance(record.msg, str) else UNSERIALIZABLE_OBJECT


def _render_exception(record: logging.LogRecord) -> str | None:
    if not isinstance(record.exc_info, tuple) or len(record.exc_info) < 2 or record.exc_info[1] is None:
        return None
    try:
        return record.exc_text or SecretRedactionFilter._formatter.formatException(record.exc_info)
    except Exception:
        return "REDACTED"


def _process_record(record: logging.LogRecord, *, base64_limit: int, text_limit: int, redact: bool) -> bool:
    if _is_redacted(record):
        return True
    message: Final = _render_message(record)
    exception: Final = _render_exception(record)
    stack: Final = record.stack_info if isinstance(record.stack_info, str) else None
    substituted_color: Final = _substituted_color_message(record)
    extras: Final = (
        tuple(
            (
                key,
                value,
                _extra_structure(
                    key, substituted_color if key == "color_message" and substituted_color is not None else value
                ),
            )
            for key, value in _record_attributes(record).items()
            if key not in _STANDARD_RECORD_ATTRS
            and key != _REDACTED_RECORD_ATTR
            and not isinstance(value, _UNREDACTED_SCALAR_TYPES)
        )
        if redact
        else ()
    )
    extra_leaves: Final = tuple(
        itertools.chain.from_iterable(_string_leaves(key, prepared) for key, _, prepared in extras)
    )
    raw_template: Final = record.msg if redact and isinstance(record.msg, str) and record.args else None
    color_template: Final = record.__dict__.get("color_message")
    raw_color: Final = color_template if redact and isinstance(color_template, str) and record.args else None
    leaves: Final = (
        extra_leaves
        + (((None, raw_template),) if raw_template is not None else ())
        + (((None, raw_color),) if raw_color is not None else ())
    )
    candidate: Final = diagnostics.run(
        lambda native: native.process_diagnostic(message, exception, stack, leaves, (redact, base64_limit, text_limit)),
        lambda: _python_process_diagnostic(message, exception, stack, leaves, redact, base64_limit, text_limit),
    )
    processed_message, processed_exception, processed_stack, processed_leaves, _ = (
        candidate
        if len(candidate[3]) == len(leaves)
        else _python_process_diagnostic(message, exception, stack, leaves, redact, base64_limit, text_limit)
    )
    raw_template_changed: Final = raw_template is not None and processed_leaves[len(extra_leaves)] != raw_template
    safe_message: Final = "REDACTED" if raw_template_changed and processed_message == message else processed_message
    if redact or safe_message != message:
        record.msg = safe_message  # rebind-ok: the Filter interface mutates the record
        record.args = None  # rebind-ok: the rendered message replaces interpolation inputs
    if processed_exception is not None:
        record.exc_text = processed_exception  # rebind-ok: the Filter interface mutates the record
    if processed_stack is not None:
        record.stack_info = processed_stack  # rebind-ok: the Filter interface mutates the record
    processed_values: Final = iter(processed_leaves[: len(extra_leaves)])
    for key, original, prepared in extras:
        replacement = _sort_processed_sets(original, _replace_string_leaves(prepared, processed_values))
        if not _scrubbing_changed_nothing(replacement, original):
            setattr(record, key, replacement)
    raw_color_changed: Final = (
        raw_color is not None and processed_leaves[len(extra_leaves) + int(raw_template is not None)] != raw_color
    )
    if raw_color_changed and getattr(record, "color_message", None) == substituted_color:
        setattr(record, "color_message", "REDACTED")
    setattr(record, _REDACTED_RECORD_ATTR, _REDACTED_STAMP)
    return True


def _redact_json_record(value: object) -> object:
    prepared: Final = safe_json_structure(value)
    leaves: Final = tuple(_string_leaves(None, prepared))
    candidate: Final = diagnostics.run(
        lambda native: native.process_diagnostic("", None, None, leaves, (True, 0, 0))[3],
        lambda: tuple(python_redact_structured_value(key, text) for key, text in leaves),
    )
    replacements: Final = (
        candidate
        if len(candidate) == len(leaves)
        else tuple(python_redact_structured_value(key, text) for key, text in leaves)
    )
    return _sort_processed_sets(value, _replace_string_leaves(prepared, iter(replacements)))


class StdoutLogTruncationFilter(logging.Filter):
    """Bounds how much of an oversized log line reaches stdout.

    A provider error string can echo the whole request payload, so one failed agentic
    request writes hundreds of KB to stdout, repeatedly as the exception propagates from
    the router to the proxy handler and into its traceback, all inline on the event loop.

    At every level, in the message and in the traceback alike, a base64 run longer than
    MAX_BASE64_LENGTH_STDOUT_LOG collapses to a size placeholder first: a multi-megabyte
    document upload otherwise costs seconds of event-loop time per DEBUG line in the
    secret regex alone. Hex and decimal runs (digests, numeric ids) are left alone unless
    they are one repeated character, which is what a zero-filled payload encodes to.
    The text around a run stays, since dumping payloads is the point of
    `--detailed_debug`, and logging callbacks (OTEL, Datadog, etc.) don't run through
    logging filters at all, so they still get the untouched record.
    """

    _formatter = logging.Formatter()

    def filter(self, record: logging.LogRecord) -> bool:
        try:
            message: Final = record.getMessage()
        except (TypeError, ValueError):
            return True

        base64_limit: Final = _get_max_base64_length_stdout_log()
        collapsed: Final = _collapse_base64_runs(message, base64_limit) if base64_limit > 0 else message
        limit: Final = _get_max_string_length_stdout_log() if record.levelno >= logging.INFO else 0
        bounded: Final = _truncate_for_stdout_log(collapsed, limit) if 0 < limit < len(collapsed) else collapsed
        if bounded != message:
            record.msg = bounded  # rebind-ok: the Filter interface mutates the record
            record.args = None  # rebind-ok: args are consumed by the rewritten message above

        if not isinstance(record.exc_info, tuple):
            return True

        exc_text: Final = record.exc_text or self._formatter.formatException(record.exc_info)
        collapsed_exc: Final = _collapse_base64_runs(exc_text, base64_limit) if base64_limit > 0 else exc_text
        bounded_exc: Final = (
            _truncate_for_stdout_log(collapsed_exc, limit) if 0 < limit < len(collapsed_exc) else collapsed_exc
        )
        if bounded_exc != exc_text:
            record.exc_text = bounded_exc  # rebind-ok: the Filter interface mutates the record

        return True


class DiagnosticProcessingFilter(StdoutLogTruncationFilter):
    def filter(self, record: logging.LogRecord) -> bool:
        return _process_record(
            record,
            base64_limit=_get_max_base64_length_stdout_log(),
            text_limit=_get_max_string_length_stdout_log() if record.levelno >= logging.INFO else 0,
            redact=_ENABLE_SECRET_REDACTION,
        )


_diagnostic_filter: Final = DiagnosticProcessingFilter()


class CorrelationContextFilter(logging.Filter):
    """Stamps each log record with the current request's trace_id and session_id from contextvars.

    Works in tandem with JsonFormatter: the formatter's record.__dict__ loop picks up these
    attributes as first-class JSON fields without any formatter-level code.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if not litellm.request_correlation_in_logs:
            return True
        trace_id: Final = trace_id_var.get()
        if trace_id:
            record.trace_id = trace_id  # rebind-ok: stamping the LogRecord is the Filter interface's contract
        session_id: Final = session_id_var.get()
        if session_id:
            record.session_id = session_id  # rebind-ok: stamping the LogRecord is the Filter interface's contract
        return True


_correlation_filter: Final = CorrelationContextFilter()


_LOG_FORMAT_PREFIX: Final = "%(asctime)s - %(name)s:%(levelname)s"
_LOG_FORMAT_SUFFIX: Final = ": %(filename)s:%(lineno)s - %(message)s"
_PLAIN_LOG_FORMAT: Final = _LOG_FORMAT_PREFIX + _LOG_FORMAT_SUFFIX
_COLOR_LOG_FORMAT: Final = f"\033[92m{_LOG_FORMAT_PREFIX}\033[0m{_LOG_FORMAT_SUFFIX}"


def _stream_is_tty(stream: TextIO | None) -> bool:
    """True when the stream is an open interactive terminal; never raises.

    A stream can be None (pythonw/embedded interpreters), lack isatty entirely
    (GUI log-redirect shims), or be closed; import must survive all three.
    """
    try:
        return stream is not None and stream.isatty()
    except (AttributeError, ValueError):
        return False


def _plain_log_format(stdout: TextIO | None, stderr: TextIO | None) -> str:
    """The plain-text log format, colorized only when both streams are an interactive terminal.

    Honors the NO_COLOR convention from no-color.org: color is disabled when
    NO_COLOR is present with a non-empty value.
    """
    if os.environ.get("NO_COLOR"):
        return _PLAIN_LOG_FORMAT
    return _COLOR_LOG_FORMAT if _stream_is_tty(stdout) and _stream_is_tty(stderr) else _PLAIN_LOG_FORMAT


class LevelRoutingStreamHandler(logging.StreamHandler):
    """Writes records below WARNING and invalid-key warnings to stdout, others to stderr.

    Collectors that derive severity from the stream report every stderr line as an error.
    Invalid-key warnings route to stdout so LITELLM_LOG=ERROR can suppress them.
    """

    def emit(self, record: logging.LogRecord) -> None:
        is_stdout_record: Final = record.levelno < logging.WARNING or (
            record.levelno == logging.WARNING and record.name == verbose_proxy_stdout_logger.name
        )
        preferred: Final = sys.stdout if is_stdout_record else sys.stderr
        if preferred is None or getattr(preferred, "closed", False):
            self.stream = sys.stderr
        else:
            self.stream = preferred
        super().emit(record)


def _parse_json_logs_env(value: str | None) -> bool:
    """Strict opt-in parse for the JSON_LOGS env var: only "true" (any case) enables JSON logs.

    Matches the reader in litellm-proxy-extras/_logging.py. The previous
    bool(os.getenv(...)) treated any non-empty value, including "false" and "0",
    as enabled.
    """
    return (value or "").lower() == "true"


def resolve_log_level(log_level: str) -> int:
    return getattr(logging, log_level.upper())


json_logs: Final = _parse_json_logs_env(os.getenv("JSON_LOGS"))
# Create a handler for the logger (you may need to adapt this based on your needs)
log_level: Final = os.getenv("LITELLM_LOG", "DEBUG")
numeric_level: Final[int] = resolve_log_level(log_level)
handler: Final = LevelRoutingStreamHandler()
handler.setLevel(numeric_level)
handler.addFilter(_secret_filter)
handler.addFilter(_correlation_filter)


def _try_parse_json_message(message: str) -> dict[str, object] | None:
    """
    Try to parse a log message as JSON. Returns parsed dict if valid, else None.
    Handles messages that are entirely valid JSON (e.g. json.dumps output).
    Uses shared safe_json_loads for consistent error handling.
    """
    if not message or not isinstance(message, str):
        return None
    msg_stripped: Final = message.strip()
    if not (msg_stripped.startswith("{") or msg_stripped.startswith("[")):
        return None
    parsed: Final = _DECODED_LOG_PAYLOAD.validate_python(safe_json_loads(message, default=None))
    if parsed is None or not isinstance(parsed, dict):
        return None
    return parsed


def _try_parse_embedded_python_dict(message: str) -> dict[str, object] | None:
    """
    Try to find and parse a Python dict repr (e.g. str(d) or repr(d)) embedded in
    the message. Handles patterns like:
    "get_available_deployment for model: X, Selected deployment: {'model_name': '...', ...} for model: X"
    Uses ast.literal_eval for safe parsing. Returns the parsed dict or None.
    """
    if not message or not isinstance(message, str) or "{" not in message:
        return None
    i = 0
    while i < len(message):
        start = message.find("{", i)
        if start == -1:
            break
        depth = 0
        for j in range(start, len(message)):
            c = message[j]
            if c == "{":
                depth += 1
            elif c == "}":
                depth -= 1
                if depth == 0:
                    substr = message[start : j + 1]
                    try:
                        result: object = _DECODED_LOG_PAYLOAD.validate_python(ast.literal_eval(substr))
                        if isinstance(result, dict) and len(result) > 0:
                            return result
                    except (ValueError, SyntaxError, TypeError):
                        pass
                    break
        i = start + 1
    return None


# Standard LogRecord attribute names - used to identify 'extra' fields.
# Derived at runtime so we automatically include version-specific attrs (e.g. taskName).
def _get_standard_record_attrs() -> frozenset:
    """Standard LogRecord attribute names - excludes extra keys from logger.debug(..., extra={...})."""
    return frozenset(logging.LogRecord("", 0, "", 0, "", (), None).__dict__.keys())


_STANDARD_RECORD_ATTRS: Final = _get_standard_record_attrs()
_NON_EXTRA_RECORD_ATTRS: Final = _STANDARD_RECORD_ATTRS | {_REDACTED_RECORD_ATTR}

# CorrelationContextFilter is the only legitimate source for these two JSON fields;
# see JsonFormatter.format() for why they're excluded from the generic message-content
# and extra-attribute promotion paths.
_RESERVED_CORRELATION_FIELDS: Final = frozenset(("trace_id", "session_id"))


class JsonFormatter(Formatter):
    def __init__(self):
        super().__init__()

    def formatTime(self, record, datefmt=None):
        # Use datetime to format the timestamp in ISO 8601 format
        dt: Final = datetime.fromtimestamp(record.created)
        return dt.isoformat()

    def format(self, record):
        message_str: Final = record.getMessage()
        json_record: Final[dict[str, object]] = {
            "message": message_str,
            "level": record.levelname,
            "timestamp": self.formatTime(record),
        }

        # Parse embedded JSON or Python dict repr in message so sub-fields become first-class properties.
        # trace_id/session_id are excluded here unconditionally (not just "if not already
        # set") - CorrelationContextFilter is the only legitimate source for these two
        # fields, and a message that merely happens to parse as JSON/dict (e.g. a proxy
        # log line dumping raw request headers) must never be able to claim them, even on
        # a record the filter hasn't stamped yet (no correlation context active for it).
        parsed = _try_parse_json_message(message_str)
        if parsed is None:
            parsed = _try_parse_embedded_python_dict(message_str)
        if parsed is not None:
            for key, value in parsed.items():
                if key not in json_record and key not in _RESERVED_CORRELATION_FIELDS:
                    json_record[key] = value

        # Include extra attributes passed via logger.debug("msg", extra={...})
        for key, value in _record_attributes(record).items():
            if key not in _NON_EXTRA_RECORD_ATTRS and key not in json_record:
                json_record[key] = value

        # trace_id/session_id are reserved: CorrelationContextFilter is the only
        # legitimate source for these two fields. Without this, a message string
        # that happens to parse as JSON/dict (e.g. a proxy log line dumping raw
        # request headers) with a "trace_id"/"session_id" key would have already
        # claimed the key at the parsed-message step above, and the extra-attributes
        # loop's "key not in json_record" guard would then skip the real value -
        # letting a caller-supplied header spoof another request's correlation ids.
        for reserved_key in _RESERVED_CORRELATION_FIELDS:
            value = getattr(record, reserved_key, None)
            if value:
                json_record[reserved_key] = value

        # Set component/logger only if not already supplied via extra={...}
        if "component" not in json_record:
            json_record["component"] = record.name
        if "logger" not in json_record:
            json_record["logger"] = f"{record.filename}:{record.lineno}"

        if record.exc_info:
            json_record["stacktrace"] = record.exc_text or self.formatException(record.exc_info)

        return safe_dumps(
            json_record if _is_redacted(record) or not _ENABLE_SECRET_REDACTION else _redact_json_record(json_record)
        )


class CorrelationPlainFormatter(logging.Formatter):
    """Appends trace_id/session_id to plain-text log lines stamped by CorrelationContextFilter.

    Mirrors JsonFormatter's handling of these two fields so request_correlation_in_logs
    behaves the same whether or not json_logs is enabled.
    """

    def format(self, record: logging.LogRecord) -> str:
        rendered: Final = super().format(record)
        formatted: Final = rendered if _is_redacted(record) else redact_string(rendered)
        trace_id: Final = getattr(record, "trace_id", None)
        session_id: Final = getattr(record, "session_id", None)
        if not trace_id and not session_id:
            return formatted
        parts: Final = tuple(
            p
            for p in (f"trace_id={trace_id}" if trace_id else None, f"session_id={session_id}" if session_id else None)
            if p
        )
        return f"{formatted} [{' '.join(parts)}]"


# Function to set up exception handlers for JSON logging
def _setup_json_exception_handlers(formatter):
    # Create a handler with JSON formatting for exceptions
    error_handler: Final = logging.StreamHandler()
    error_handler.setFormatter(formatter)
    error_handler.addFilter(_diagnostic_filter)
    error_handler.addFilter(_secret_filter)
    error_handler.addFilter(_correlation_filter)

    # Setup excepthook for uncaught exceptions
    def json_excepthook(exc_type, exc_value, exc_traceback):
        record: Final = logging.LogRecord(
            name="LiteLLM",
            level=logging.ERROR,
            pathname="",
            lineno=0,
            msg=str(exc_value),
            args=(),
            exc_info=(exc_type, exc_value, exc_traceback),
        )
        error_handler.handle(record)

    sys.excepthook = json_excepthook

    # Configure asyncio exception handler if possible
    try:
        import asyncio

        def async_json_exception_handler(loop, context):
            exception: Final = context.get("exception")
            if exception:
                exc_type: Final = type(exception)
                record: Final = logging.LogRecord(
                    name="LiteLLM",
                    level=logging.ERROR,
                    pathname="",
                    lineno=0,
                    msg=str(exception),
                    args=(),
                    exc_info=(exc_type, exception, exception.__traceback__),
                )
                error_handler.handle(record)
            else:
                loop.default_exception_handler(context)

        asyncio.get_event_loop().set_exception_handler(async_json_exception_handler)
    except Exception:
        pass


# Create a formatter and set it for the handler
if json_logs:
    handler.setFormatter(JsonFormatter())
    _setup_json_exception_handlers(JsonFormatter())
else:
    formatter: Final = CorrelationPlainFormatter(
        _plain_log_format(sys.stdout, sys.stderr),
        datefmt="%H:%M:%S",
    )

    handler.setFormatter(formatter)

verbose_proxy_logger = logging.getLogger("LiteLLM Proxy")
# Malformed virtual key rejections log through this child; LevelRoutingStreamHandler
# writes its WARNING records to stdout. It has no handler or level of its own.
verbose_proxy_stdout_logger: Final = verbose_proxy_logger.getChild("stdout")
verbose_router_logger = logging.getLogger("LiteLLM Router")
verbose_logger = logging.getLogger("LiteLLM")

# Add the handler to the loggers
verbose_router_logger.addHandler(handler)
verbose_proxy_logger.addHandler(handler)
verbose_logger.addHandler(handler)

# Filters attached to the logger, not the handler, survive callers swapping in their own
# handlers (JSON mode, uvicorn log config, a host app's root handler).
verbose_router_logger.addFilter(_diagnostic_filter)
verbose_proxy_logger.addFilter(_diagnostic_filter)
verbose_proxy_stdout_logger.addFilter(_diagnostic_filter)
verbose_logger.addFilter(_diagnostic_filter)

_ANSI_ESCAPE: Final = re.compile(r"\x1b\[[0-9;]*m")


def print_startup_message(message: str) -> None:
    """Write a boot-time message meant for the operator's terminal.

    Plain mode prints it unchanged, colors and spacing included. With JSON logs on, every line of
    stdout is read by a log pipeline, so the message goes out as one INFO record of the proxy logger
    instead, on stdout like the print whatever handlers the logging config gave that logger, with the
    ANSI color codes and surrounding blank lines removed, located at the caller and emitted at any log
    level, whole and unredacted as the print showed it; a blank spacer produces no record.
    """
    if not litellm.json_logs:
        print(message)  # noqa: T201  # plain mode: this is the operator-terminal print it replaces
        return
    text: Final = _ANSI_ESCAPE.sub("", message).strip("\n")
    if not text:
        return
    frame: Final = inspect.currentframe()
    caller: Final = frame.f_back if frame is not None else None
    record: Final = verbose_proxy_logger.makeRecord(
        verbose_proxy_logger.name,
        logging.INFO,
        caller.f_code.co_filename if caller is not None else __file__,
        caller.f_lineno if caller is not None else 0,
        text,
        (),
        None,
        func=caller.f_code.co_name if caller is not None else None,
    )
    if not _process_record(record, base64_limit=_get_max_base64_length_stdout_log(), text_limit=0, redact=False):
        return
    _startup_json_handler().handle(record)


def _startup_json_handler() -> logging.Handler:
    """The proxy's JSON handler on stdout, filters included, for one startup message."""
    handler: Final = LevelRoutingStreamHandler()
    handler.setFormatter(JsonFormatter())
    handler.addFilter(_secret_filter)
    handler.addFilter(_correlation_filter)
    return handler


def _suppress_loggers():
    """Suppress noisy loggers at INFO level"""
    # Suppress httpx request logging at INFO level
    httpx_logger: Final = logging.getLogger("httpx")
    httpx_logger.setLevel(logging.WARNING)

    # Suppress APScheduler logging at INFO level
    apscheduler_executors_logger: Final = logging.getLogger("apscheduler.executors.default")
    apscheduler_executors_logger.setLevel(logging.WARNING)
    apscheduler_scheduler_logger: Final = logging.getLogger("apscheduler.scheduler")
    apscheduler_scheduler_logger.setLevel(logging.WARNING)


_REDACTED_THIRD_PARTY_LOGGERS: Final[tuple[str, ...]] = (
    "apscheduler.executors.default",
    "apscheduler.scheduler",
    "asyncio",
    "backoff",
    "httpx",
    "uvicorn.error",
)

# Access loggers, which emit the full request target, so a credential passed as a
# query parameter (e.g. `/key/info?key=`) lands on stdout verbatim. uvicorn.access
# covers uvicorn.run, --run_gunicorn (its worker_class is UvicornWorker, so the
# access line is still uvicorn's) and an embedding host app. --run_hypercorn and
# --run_granian log through their own loggers in their own record shapes, and
# both ship with access logging off.
_REDACTED_ACCESS_LOGGERS: Final[tuple[str, ...]] = ("uvicorn.access",)


def _redact_third_party_loggers() -> None:
    """Extend secret redaction to records litellm does not emit directly.

    litellm's own loggers are covered by the filter on their shared handler, but a
    litellm value can also reach a log record through a dependency that logs on its
    own logger. Those records never pass through a litellm handler.

    The filter is attached to each emitting logger rather than to the root logger or
    to root's handlers. `Logger.handle` applies the emitting logger's filters before
    any handler runs, so redaction happens once, at the earliest point in the
    record's life, and covers every downstream handler regardless of who owns it.
    The alternatives do not hold: `callHandlers` consults ancestors for handlers but
    never for filters, so a filter on the root logger never sees these records at
    all, and a filter on a root handler only covers that one handler, leaving
    handlers registered earlier or on the emitting logger itself untouched.

    Each name is the exact logger a dependency emits on; a parent name would not
    cover its children, for the same reason the root logger does not.
    """
    for name in _REDACTED_THIRD_PARTY_LOGGERS:
        logging.getLogger(name).addFilter(_secret_filter)
    for name in _REDACTED_ACCESS_LOGGERS:
        logging.getLogger(name).addFilter(_access_log_path_filter)
        logging.getLogger(name).addFilter(_access_log_filter)


# Call the suppression function
_suppress_loggers()
_redact_third_party_loggers()

ALL_LOGGERS: Final = [
    logging.getLogger(),
    verbose_logger,
    verbose_router_logger,
    verbose_proxy_logger,
]


def _get_loggers_to_initialize():
    """
    Get all loggers that should be initialized with the JSON handler.

    Includes third-party integration loggers (like langfuse) if they are
    configured as callbacks.
    """
    import litellm

    loggers: Final = list(ALL_LOGGERS)

    # Add langfuse logger if langfuse is being used as a callback
    langfuse_callbacks: Final = {"langfuse", "langfuse_otel"}
    all_callbacks: Final = set(litellm.success_callback + litellm.failure_callback)
    if langfuse_callbacks & all_callbacks:
        loggers.append(logging.getLogger("langfuse"))

    return loggers


def _initialize_loggers_with_handler(handler: logging.Handler):
    """
    Initialize all loggers with a handler

    - Adds a handler to each logger
    - Prevents bubbling to parent/root (critical to prevent duplicate JSON logs)
    """
    handler.addFilter(_secret_filter)
    handler.addFilter(_correlation_filter)
    for lg in _get_loggers_to_initialize():
        lg.handlers.clear()  # remove any existing handlers
        lg.addHandler(handler)  # add JSON formatter handler
        lg.propagate = False  # prevent bubbling to parent/root
    if (configured_level := os.getenv("LITELLM_LOG")) is not None:
        logging.getLogger().setLevel(resolve_log_level(configured_level))


def _get_uvicorn_json_log_config():
    """
    Generate a uvicorn log_config dictionary that applies JSON formatting to all loggers.

    This ensures that uvicorn's access logs, error logs, and all application logs
    are formatted as JSON when json_logs is enabled.
    """
    json_formatter_class: Final = "litellm._logging.JsonFormatter"

    # Use the module-level log_level variable for consistency
    uvicorn_log_level: Final = log_level.upper()

    log_config: Final = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "json": {
                "()": json_formatter_class,
            },
            "default": {
                "()": json_formatter_class,
            },
            "access": {
                "()": json_formatter_class,
            },
        },
        "handlers": {
            "default": {
                "formatter": "json",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            },
            "access": {
                "formatter": "access",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            },
        },
        "loggers": {
            "uvicorn": {
                "handlers": ["default"],
                "level": uvicorn_log_level,
                "propagate": False,
            },
            "uvicorn.error": {
                "handlers": ["default"],
                "level": uvicorn_log_level,
                "propagate": False,
            },
            "uvicorn.access": {
                "handlers": ["access"],
                "level": uvicorn_log_level,
                "propagate": False,
            },
        },
    }

    return log_config


def turn_on_json() -> None:
    """
    Turn on JSON logging

    - Adds a JSON formatter to all loggers
    """
    handler: Final = LevelRoutingStreamHandler()
    handler.setLevel(numeric_level)
    handler.setFormatter(JsonFormatter())
    _initialize_loggers_with_handler(handler)
    # Set up exception handlers
    _setup_json_exception_handlers(JsonFormatter())


_turn_on_json = turn_on_json


def turn_on_debug() -> None:
    verbose_logger.setLevel(level=logging.DEBUG)  # set package log to debug
    verbose_router_logger.setLevel(level=logging.DEBUG)  # set router logs to debug
    verbose_proxy_logger.setLevel(level=logging.DEBUG)  # set proxy logs to debug


_turn_on_debug = turn_on_debug


def _disable_debugging():
    """Disable the package, router, and proxy verbose loggers."""
    verbose_logger.disabled = True
    verbose_router_logger.disabled = True
    verbose_proxy_logger.disabled = True
    verbose_proxy_stdout_logger.disabled = True


def _enable_debugging():
    verbose_logger.disabled = False
    verbose_router_logger.disabled = False
    verbose_proxy_logger.disabled = False
    verbose_proxy_stdout_logger.disabled = False


def print_verbose(print_statement: object, *args: object) -> None:
    try:
        if not set_verbose:
            return
        message: Final = str(print_statement) % args if args else str(print_statement)
        print(redact_secrets(message))  # noqa: T201
    except Exception:
        pass


def is_debugging_on() -> bool:
    """
    Returns True if debugging is on
    """
    return verbose_logger.isEnabledFor(logging.DEBUG) or set_verbose is True


_is_debugging_on = is_debugging_on