fix(logging): keep proxy-set fields over keys parsed from a JSON log message

litellm/_logging.py

@@ -3,6 +3,7 @@ import contextvars
 import functools
 import inspect
 import itertools
+import json
 import logging
 import os
 import re
@@ -115,6 +116,17 @@ def _plain_text(value: object) -> str:
         return UNSERIALIZABLE_OBJECT
 
 
+_BARE_LOGFMT_VALUE: Final = re.compile(r"[\w.:/@+-]+", re.ASCII)
+_JSON_ESCAPED_BRACES: Final = str.maketrans({"{": "\\u007b", "}": "\\u007d"})
+
+
+def logfmt_value(value: object) -> str:
+    text: Final = _plain_text(value)
+    if _BARE_LOGFMT_VALUE.fullmatch(text):
+        return text
+    return json.dumps(text).translate(_JSON_ESCAPED_BRACES)
+
+
 def redact_secrets(value: str) -> str:
     """Public API: redact known secret/credential patterns from an arbitrary string.
 
@@ -754,10 +766,14 @@ def _get_standard_record_attrs() -> frozenset:
 _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"))
+_UNPROMOTABLE_MESSAGE_KEYS: Final = frozenset(("trace_id", "session_id", "stacktrace"))
+
+
+def _parsed_message_fields(message: str) -> Mapping[str, object]:
+    parsed: Final = _try_parse_json_message(message)
+    if parsed is not None:
+        return parsed
+    return _try_parse_embedded_python_dict(message) or {}
 
 
 class JsonFormatter(Formatter):
@@ -769,54 +785,28 @@ class JsonFormatter(Formatter):
         dt: Final = datetime.fromtimestamp(record.created)
         return dt.isoformat()
 
-    def format(self, record):
+    def format(self, record: logging.LogRecord) -> str:
         message_str: Final = _ANSI_ESCAPE.sub("", record.getMessage())
-        json_record: Final[dict[str, object]] = {
-            "message": message_str,
-            "level": record.levelname,
-            "timestamp": self.formatTime(record),
+        fixed_fields: Final = {"message": message_str, "level": record.levelname, "timestamp": self.formatTime(record)}
+        extra_fields: Final = {
+            key: value for key, value in _record_attributes(record).items() if key not in _NON_EXTRA_RECORD_ATTRS
         }
-
-        # 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)
-
+        stacktrace_field: Final[Mapping[str, str]] = (
+            {"stacktrace": record.exc_text or self.formatException(record.exc_info)} if record.exc_info else {}
+        )
+        logger_fields: Final = {
+            "component": record.name,
+            "logger": f"{record.filename}:{record.lineno}",
+            **extra_fields,
+            **stacktrace_field,
+            **fixed_fields,
+        }
+        message_fields: Final = {
+            key: value
+            for key, value in _parsed_message_fields(message_str).items()
+            if key not in logger_fields and key not in _UNPROMOTABLE_MESSAGE_KEYS
+        }
+        json_record: Final = {**fixed_fields, **message_fields, **logger_fields}
         return safe_dumps(
             json_record if _is_redacted(record) or not _ENABLE_SECRET_REDACTION else _redact_json_record(json_record)
         )

litellm/proxy/common_request_processing.py

@@ -34,7 +34,7 @@ from starlette.types import Receive, Scope, Send
 from typing_extensions import Never
 
 import litellm
-from litellm._logging import redact_internal_details_from_client_message, verbose_proxy_logger
+from litellm._logging import logfmt_value, redact_internal_details_from_client_message, verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.anthropic_interface.exceptions import AnthropicErrorSseFrame, anthropic_error_sse_frame
 from litellm.constants import (
@@ -403,7 +403,7 @@ async def _record_streaming_client_disconnect_if_needed(
 
     verbose_proxy_logger.debug(
         "Recorded streaming client disconnect with error_code=499 for litellm_call_id=%s",
-        request_data.get("litellm_call_id"),
+        logfmt_value(request_data.get("litellm_call_id")),
     )
     return True
 
@@ -492,7 +492,7 @@ async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, respons
     if partial_responses_event is not None:
         verbose_proxy_logger.debug(
             "Billing partial streamed Responses API spend after client disconnect, litellm_call_id=%s",
-            sanitize_for_log(logging_obj.litellm_call_id),
+            logfmt_value(logging_obj.litellm_call_id),
         )
         return await _dispatch_disconnect_billing_event(logging_obj, partial_responses_event)
     chunks: Final[object] = getattr(response, "chunks", None)
@@ -501,7 +501,7 @@ async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, respons
     verbose_proxy_logger.debug(
         "Billing partial streamed spend for %s chunks after client disconnect, litellm_call_id=%s",
         len(chunks),
-        request_data.get("litellm_call_id"),
+        logfmt_value(request_data.get("litellm_call_id")),
     )
     messages: Final[object] = getattr(response, "messages", None)
     try:
@@ -1871,7 +1871,7 @@ def log_llm_api_exception(e: Exception, litellm_call_id: str | None) -> None:
         verbose_proxy_logger.info(
             "litellm.proxy.proxy_server._handle_llm_api_exception(): client disconnected, "
             "upstream LLM request cancelled - litellm_call_id=%s",
-            litellm_call_id,
+            logfmt_value(litellm_call_id),
             extra=MappingProxyType({"litellm_call_id": litellm_call_id}),
         )
         return
@@ -1882,7 +1882,7 @@ def log_llm_api_exception(e: Exception, litellm_call_id: str | None) -> None:
     )
     log_fn(
         "litellm.proxy.proxy_server._handle_llm_api_exception(): Exception occured - litellm_call_id=%s - %s",
-        litellm_call_id,
+        logfmt_value(litellm_call_id),
         e,
         extra=MappingProxyType({"litellm_call_id": litellm_call_id}),
     )

litellm/proxy/common_utils/token_count_log.py

@@ -1,5 +1,3 @@
-import json
-import re
 from asyncio import CancelledError
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
@@ -8,23 +6,11 @@ from typing import Final, TypeVar
 
 from fastapi import status
 
-from litellm._logging import verbose_proxy_logger
+from litellm._logging import logfmt_value, verbose_proxy_logger
 from litellm.proxy.common_utils.openai_error_payload import error_status_code
 from litellm.types.token_count import TokenCountSource
 from litellm.types.utils import TokenCountResponse
 
-_BARE_LOGFMT_VALUE: Final = re.compile(r"[\w.:/@+-]+", re.ASCII)
-_JSON_ESCAPED_BRACES: Final = str.maketrans({"{": "\\u007b", "}": "\\u007d"})
-
-
-def _logfmt_value(value: str | None) -> str:
-    if value is None:
-        return "None"
-    if _BARE_LOGFMT_VALUE.fullmatch(value):
-        return value
-    return json.dumps(value).translate(_JSON_ESCAPED_BRACES)
-
-
 _Counted: Final = TypeVar("_Counted")
 _Rejection: Final = TypeVar("_Rejection")
 _ProviderCount: Final = TypeVar("_ProviderCount", bound=Mapping[str, object])
@@ -59,10 +45,10 @@ def _log_token_count(request: _CountRequest, fields: _CountFields, status_code:
     verbose_proxy_logger.info(
         "token count - litellm_call_id=%s request_model=%s model_id=%s model_used=%s input_tokens=%s "
         "count_source=%s status_code=%s",
-        _logfmt_value(request.litellm_call_id),
-        _logfmt_value(request.request_model),
-        _logfmt_value(request.model_id),
-        _logfmt_value(fields.model_used),
+        logfmt_value(request.litellm_call_id),
+        logfmt_value(request.request_model),
+        logfmt_value(request.model_id),
+        logfmt_value(fields.model_used),
         fields.input_tokens,
         fields.count_source,
         status_code,

litellm/proxy/proxy_server.py

@@ -311,6 +311,7 @@ from litellm import Router
 from litellm._internal_context import service_target, with_service_target
 from litellm._logging import (
     STARTUP_OUTPUT_SETTINGS,
+    logfmt_value,
     print_startup_message,
     redact_string,
     verbose_proxy_logger,
@@ -9980,7 +9981,7 @@ def _restamp_streaming_chunk_model(
     if not model_mismatch_logged and downstream_model != target_model:
         verbose_proxy_logger.debug(
             "litellm_call_id=%s: streaming chunk model mismatch - target=%r downstream=%r fallback_was_attempted=%s. Overriding chunk model to target.",
-            request_data.get("litellm_call_id"),
+            logfmt_value(request_data.get("litellm_call_id")),
             target_model,
             downstream_model,
             fallback_was_attempted,
@@ -10001,7 +10002,7 @@ def _restamp_streaming_chunk_model(
     except Exception as e:
         verbose_proxy_logger.error(
             "litellm_call_id=%s: failed to override chunk.model=%r on chunk_type=%s. error=%s",
-            request_data.get("litellm_call_id"),
+            logfmt_value(request_data.get("litellm_call_id")),
             target_model,
             type(restamped_chunk),
             str(e),

tests/unit/proxy/conftest.py

@@ -3,11 +3,13 @@
 import asyncio
 import copy
 import inspect
+import json
 import logging
 import os
 import tempfile
 import warnings
 from collections.abc import Awaitable, Callable, Iterator
+from io import StringIO
 from typing import Dict, Final, Optional
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -18,7 +20,7 @@ from prisma.errors import ClientNotConnectedError
 
 import litellm
 import litellm.proxy.proxy_server
-from litellm._logging import verbose_proxy_logger
+from litellm._logging import JsonFormatter, verbose_proxy_logger
 from litellm._service_logger import ServiceTypes
 from litellm.integrations.otel.model.payloads import ServiceSpanData
 from litellm.integrations.otel.model.spans import service_span_name
@@ -451,6 +453,19 @@ def token_count_log_lines(
         )
 
 
+@pytest.fixture
+def proxy_json_log_lines() -> Iterator[Callable[[], tuple[dict[str, object], ...]]]:
+    stream: Final = StringIO()
+    handler: Final = logging.StreamHandler(stream)
+    handler.setFormatter(JsonFormatter())
+    previous_level: Final = verbose_proxy_logger.level
+    verbose_proxy_logger.addHandler(handler)
+    verbose_proxy_logger.setLevel(logging.DEBUG)
+    yield lambda: tuple(json.loads(line) for line in stream.getvalue().splitlines())
+    verbose_proxy_logger.removeHandler(handler)
+    verbose_proxy_logger.setLevel(previous_level)
+
+
 @pytest.fixture
 def postgres_span_names() -> Iterator[Callable[[], Awaitable[tuple[str, ...]]]]:
     """The ``postgres.{verb} {table}`` names OTel would render for every DB service event

tests/unit/proxy/test_common_request_processing.py

@@ -27,8 +27,10 @@ from litellm.litellm_core_utils.bug_report import (
 )
 from litellm.types.router import RouterRateLimitError, RouterRateLimitErrorBasic
 from litellm.types.utils import CachingDetails, Delta, ModelResponseStream, StreamingChoices
+from litellm._logging import verbose_proxy_logger
 from litellm.constants import (
     CLIENT_REQUESTED_MODEL_SCOPE_KEY,
+    LITELLM_CLIENT_DISCONNECT_DETAIL,
     MAX_LITELLM_CALL_ID_LENGTH,
     RETURN_RAW_MODEL_NAME_METADATA_KEY,
     STREAM_SSE_KEEPALIVE_PING_BYTES,
@@ -51,6 +53,7 @@ from litellm.proxy.common_request_processing import (
     _has_attribute_error_in_chain,
     include_guardrail_response_requested,
     is_azure_model_router_request,
+    log_llm_api_exception,
     open_sse_before_first_byte,
     resolve_litellm_call_id,
     ttft_keepalive_interval,
@@ -82,6 +85,8 @@ from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 from litellm.utils import ModelResponseListIterator
 from litellm.router_utils.add_retry_fallback_headers import prepare_response_for_header_attachment
 
+_DICT_SHAPED_CALL_ID: Final = "{'component': 'forged', 'logger': 'x', 'litellm_call_id': 'y', 'injected': 1}"
+
 
 def test_attach_guardrail_information_copies_recorded_entries_onto_model_response():
     recorded = [
@@ -8765,6 +8770,73 @@ class TestStreamingClientDisconnectBilling:
         await stream_iter.__anext__()
         return response
 
+    @pytest.mark.asyncio
+    async def test_dict_shaped_call_id_cannot_forge_fields_on_the_partial_responses_billing_log(
+        self, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]]
+    ) -> None:
+        from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
+        from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
+
+        logging_obj: Final = LiteLLMLoggingObj(
+            model="gpt-4o-mini",
+            messages=[{"role": "user", "content": "tell me a story"}],
+            stream=True,
+            call_type="aresponses",
+            start_time=datetime.datetime.now(),
+            litellm_call_id=_DICT_SHAPED_CALL_ID,
+            function_id="test-function-id",
+        )
+        stream: Final = ResponsesAPIStreamingIterator(
+            response=httpx.Response(200),
+            model="gpt-4o-mini",
+            responses_api_provider_config=OpenAIResponsesAPIConfig(),
+            logging_obj=logging_obj,
+            request_data={"model": "gpt-4o-mini", "input": "tell me a story"},
+            custom_llm_provider="openai",
+        )
+        stream._process_chunk(
+            json.dumps(
+                {
+                    "type": "response.created",
+                    "sequence_number": 0,
+                    "response": {
+                        "id": "resp_partial",
+                        "object": "response",
+                        "created_at": 1700000000.0,
+                        "model": "gpt-4o-mini",
+                        "status": "in_progress",
+                        "output": [],
+                        "parallel_tool_calls": True,
+                        "tool_choice": "auto",
+                        "tools": [],
+                    },
+                }
+            )
+        )
+        stream._process_chunk(
+            json.dumps(
+                {
+                    "type": "response.output_text.delta",
+                    "sequence_number": 1,
+                    "item_id": "msg_partial",
+                    "output_index": 0,
+                    "content_index": 0,
+                    "delta": "The codename is AZURE-FALCON-42",
+                }
+            )
+        )
+
+        await _bill_partial_streamed_spend_on_disconnect({"litellm_logging_obj": logging_obj}, stream)
+
+        (line,) = (
+            line
+            for line in proxy_json_log_lines()
+            if "Billing partial streamed Responses API spend" in str(line["message"])
+        )
+        assert line["component"] == verbose_proxy_logger.name
+        assert "litellm_call_id" not in line
+        assert "injected" not in line
+
     @pytest.mark.asyncio
     async def test_disconnect_bills_partial_streamed_spend(self):
         recorder = _RecordingSuccessLogger()
@@ -11968,6 +12040,38 @@ class TestErrorLogCarriesCallId:
         assert record.litellm_call_id == call_id
         assert call_id in record.getMessage()
 
+    @pytest.mark.parametrize(
+        "exc",
+        [
+            pytest.param(HTTPException(status_code=400, detail={"error": "Invalid model name"}), id="error_line"),
+            pytest.param(
+                HTTPException(status_code=499, detail=LITELLM_CLIENT_DISCONNECT_DETAIL), id="client_disconnect_line"
+            ),
+        ],
+    )
+    def test_dict_shaped_call_id_cannot_forge_json_log_fields(
+        self, exc: HTTPException, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]]
+    ) -> None:
+        log_llm_api_exception(exc, _DICT_SHAPED_CALL_ID)
+
+        (line,) = proxy_json_log_lines()
+        assert line["component"] == verbose_proxy_logger.name
+        assert str(line["logger"]).startswith("common_request_processing.py:")
+        assert line["litellm_call_id"] == _DICT_SHAPED_CALL_ID
+        assert "injected" not in line
+
+    def test_provider_error_json_in_the_exception_still_becomes_a_field_next_to_a_dict_shaped_call_id(
+        self, proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]]
+    ) -> None:
+        provider_error: Final = "Invalid model name passed in model=nope-model"
+
+        log_llm_api_exception(
+            HTTPException(status_code=400, detail={"error": provider_error}), "{'error': 'forged', 'injected': 1}"
+        )
+
+        (line,) = proxy_json_log_lines()
+        assert line["error"] == provider_error
+
 
 class TestAnthropicMessagesStreamErrorFrame:
     """A ``/v1/messages`` stream that fails after the headers are out has to say so with an

tests/unit/proxy/test_response_model_sanitization.py

@@ -1,11 +1,13 @@
 import asyncio
 import json
-from typing import AsyncGenerator
+from collections.abc import Callable
+from typing import AsyncGenerator, Final
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 import yaml
 from fastapi.testclient import TestClient
+from pydantic import BaseModel
 
 
 import litellm
@@ -502,6 +504,34 @@ def test_restamp_streaming_chunk_rewrites_nested_response_model():
     assert chunk.response.model == "hosted_vllm/some-model"
 
 
+class _UncopyableChunk(BaseModel):
+    model: str
+
+    def model_copy(self, *, update: object = None, deep: bool = False) -> "_UncopyableChunk":
+        raise RuntimeError("chunk cannot be copied")
+
+
+def test_a_failed_restamp_logs_a_dict_shaped_call_id_without_forging_fields(
+    proxy_json_log_lines: Callable[[], tuple[dict[str, object], ...]],
+):
+    from litellm._logging import verbose_proxy_logger
+    from litellm.proxy.proxy_server import _restamp_streaming_chunk_model
+
+    chunk: Final = _UncopyableChunk(model="hosted_vllm/some-model")
+    result, _ = _restamp_streaming_chunk_model(
+        chunk=chunk,
+        requested_model_from_client="client-model",
+        request_data={"litellm_call_id": "{'component': 'forged', 'litellm_call_id': 'y', 'injected': 1}"},
+        model_mismatch_logged=True,
+    )
+
+    assert result is chunk
+    (line,) = (line for line in proxy_json_log_lines() if "failed to override chunk.model" in str(line["message"]))
+    assert line["component"] == verbose_proxy_logger.name
+    assert "litellm_call_id" not in line
+    assert "injected" not in line
+
+
 def _make_response_completed_event(model: str):
     from litellm.types.llms.openai import (
         ResponseCompletedEvent,

tests/unit/test_logging.py

@@ -555,6 +555,36 @@ def test_trace_id_and_session_id_cannot_be_injected_with_no_active_context(monke
     assert "session_id" not in cap.records[0]
 
 
+def test_message_dict_keys_never_override_fields_the_logger_sets():
+    lg, cap = _make_capture_logger("test.message_dict_precedence")
+    embedded: Final = {
+        "component": "forged-component",
+        "logger": "forged.py:1",
+        "stacktrace": "forged stacktrace",
+        "litellm_call_id": "forged-call-id",
+        "model_name": "gpt-x",
+    }
+    lg.handle(
+        lg.makeRecord(
+            lg.name,
+            logging.INFO,
+            "/srv/litellm/router.py",
+            42,
+            f"Selected deployment: {embedded} for model: gpt-x",
+            (),
+            None,
+            extra={"litellm_call_id": "real-call-id"},
+        )
+    )
+
+    record: Final = cap.records[0]
+    assert record["component"] == "test.message_dict_precedence"
+    assert record["logger"] == "router.py:42"
+    assert record["litellm_call_id"] == "real-call-id"
+    assert "stacktrace" not in record
+    assert record["model_name"] == "gpt-x"
+
+
 def test_trace_id_and_session_id_are_redacted_when_credential_shaped(monkeypatch):
     """A caller-controlled trace_id/session_id (e.g. from x-litellm-trace-id or a W3C
     baggage header) that happens to look like a real credential must not reach log