import asyncio
import datetime
import json
from datetime import timezone
from collections.abc import Mapping
from typing import Any, Final, cast
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from typing_extensions import ReadOnly, TypedDict

import litellm
from litellm.constants import (
    LITELLM_TRUNCATED_PAYLOAD_FIELD,
    LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE,
    REDACTED_BY_LITELM_STRING,
)
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy.spend_tracking.spend_tracking_utils import (
    _get_messages_for_spend_logs_payload,
    _get_proxy_server_request_for_spend_logs_payload,
    _get_request_duration_ms,
    _get_response_for_spend_logs_payload,
    _get_spend_logs_metadata,
    _get_vector_store_request_for_spend_logs_payload,
    _is_master_key,
    _redact_logged_api_key,
    _redact_prompt_leaks_in_error_string,
    _sanitize_error_information_for_spend_logs,
    _sanitize_guardrail_information_for_spend_logs,
    _sanitize_request_body_for_spend_logs_payload,
    _should_store_prompts_and_responses_in_spend_logs,
    get_logging_payload,
    get_spend_logs_id,
)
from litellm.proxy.utils import hash_token
from litellm.types.utils import (
    StandardLoggingHiddenParams,
    StandardLoggingMetadata,
    StandardLoggingModelInformation,
    StandardLoggingPayload,
)


def _get_additional_usage_values_for_usage(usage: litellm.Usage) -> dict:
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
        },
        response_obj=litellm.ModelResponse(
            id="chatcmpl-test",
            choices=[],
            usage=usage,
        ),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    return metadata["additional_usage_values"]


def test_get_logging_payload_maps_openai_cached_tokens_to_cache_read_input_tokens():
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=10,
            completion_tokens=2,
            total_tokens=12,
            prompt_tokens_details={"cached_tokens": 123},
        )
    )

    assert additional_usage_values["cache_read_input_tokens"] == 123
    assert additional_usage_values["prompt_tokens_details"]["cached_tokens"] == 123


def test_get_logging_payload_preserves_anthropic_cache_read_input_tokens():
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=10,
            completion_tokens=2,
            total_tokens=12,
            prompt_tokens_details={"cached_tokens": 123},
            cache_read_input_tokens=456,
        )
    )

    assert additional_usage_values["cache_read_input_tokens"] == 456


@pytest.mark.parametrize(
    "prompt_tokens_details",
    [None, {"cached_tokens": 0}],
)
def test_get_logging_payload_does_not_map_missing_or_zero_cached_tokens(prompt_tokens_details):
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=10,
            completion_tokens=2,
            total_tokens=12,
            prompt_tokens_details=prompt_tokens_details,
        )
    )

    assert "cache_read_input_tokens" not in additional_usage_values


def test_get_logging_payload_maps_openai_cache_write_tokens_to_cache_creation_input_tokens():
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=1000,
            completion_tokens=2,
            total_tokens=1002,
            prompt_tokens_details={"cached_tokens": 0, "cache_write_tokens": 800},
        )
    )

    assert additional_usage_values["cache_creation_input_tokens"] == 800
    assert additional_usage_values["prompt_tokens_details"]["cache_write_tokens"] == 800


def test_get_logging_payload_maps_nested_cache_creation_input_tokens():
    """
    Regression (LIT-5757): DashScope nests cache_creation_input_tokens inside
    prompt_tokens_details; SpendLogs must record it as cache_creation_input_tokens.
    """
    additional_usage_values: Final = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=2059,
            completion_tokens=31,
            total_tokens=2090,
            prompt_tokens_details={
                "cached_tokens": 0,
                "text_tokens": 2059,
                "cache_type": "ephemeral",
                "cache_creation_input_tokens": 2048,
                "cache_creation": {"ephemeral_5m_input_tokens": 2048},
            },
        )
    )

    assert additional_usage_values["cache_creation_input_tokens"] == 2048
    assert additional_usage_values["prompt_tokens_details"]["cache_write_tokens"] == 2048


def test_get_logging_payload_preserves_anthropic_cache_creation_input_tokens():
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=1000,
            completion_tokens=2,
            total_tokens=1002,
            cache_creation_input_tokens=300,
        )
    )

    assert additional_usage_values["cache_creation_input_tokens"] == 300


@pytest.mark.parametrize(
    "prompt_tokens_details",
    [None, {"cached_tokens": 100}, {"cached_tokens": 100, "cache_write_tokens": 0}],
)
def test_get_logging_payload_does_not_map_missing_or_zero_cache_write_tokens(prompt_tokens_details):
    additional_usage_values = _get_additional_usage_values_for_usage(
        litellm.Usage(
            prompt_tokens=10,
            completion_tokens=2,
            total_tokens=12,
            prompt_tokens_details=prompt_tokens_details,
        )
    )

    assert "cache_creation_input_tokens" not in additional_usage_values


def _make_standard_logging_payload_with_usage_object(usage_object: dict) -> StandardLoggingPayload:
    return StandardLoggingPayload(
        id="test-id-responses",
        call_type="responses",
        stream=False,
        response_cost=0.02,
        status="success",
        total_tokens=1010,
        prompt_tokens=1000,
        completion_tokens=10,
        startTime=1234567890.0,
        endTime=1234567891.0,
        completionStartTime=None,
        model_map_information=StandardLoggingModelInformation(model_map_key="gpt-5.6", model_map_value=None),
        model="gpt-5.6",
        model_id="model-123",
        model_group="openai",
        custom_llm_provider="openai",
        api_base="https://api.openai.com",
        metadata=StandardLoggingMetadata(
            user_api_key_hash="test_hash",
            user_api_key_alias=None,
            user_api_key_team_id=None,
            user_api_key_org_id=None,
            user_api_key_user_id=None,
            user_api_key_team_alias=None,
            spend_logs_metadata=None,
            requester_ip_address=None,
            requester_metadata=None,
            user_api_key_end_user_id=None,
            usage_object=usage_object,
        ),
        cache_hit=False,
        cache_key=None,
        saved_cache_cost=0.0,
        request_tags=[],
        end_user=None,
        requester_ip_address=None,
        messages=[],
        response={},
        error_str=None,
        model_parameters={},
        hidden_params=StandardLoggingHiddenParams(
            model_id="model-123",
            cache_key=None,
            api_base="https://api.openai.com",
            response_cost="0.02",
            litellm_overhead_time_ms=None,
            additional_headers=None,
            batch_models=None,
            litellm_model_name=None,
            usage_object=None,
        ),
    )


def test_get_logging_payload_maps_responses_api_cache_write_tokens_from_usage_object():
    """Responses API (/v1/responses) usage is not chat-Usage-shaped, so
    additional_usage_values can't derive cache tokens from response_obj.usage.
    The Admin UI Logs "Cache Creation Tokens" row reads
    additional_usage_values.cache_creation_input_tokens, so it must be filled
    from the normalized standard_logging usage_object (LIT-4633)."""
    standard_logging_payload = _make_standard_logging_payload_with_usage_object(
        usage_object={
            "prompt_tokens": 1000,
            "completion_tokens": 10,
            "total_tokens": 1010,
            "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 800, "cache_creation_tokens": 800},
        }
    )
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-5.6",
            "call_type": "responses",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
            "standard_logging_object": standard_logging_payload,
        },
        response_obj={
            "id": "resp-test",
            "usage": {
                "input_tokens": 1000,
                "output_tokens": 10,
                "total_tokens": 1010,
                "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 800},
            },
        },
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    additional_usage_values = json.loads(payload["metadata"])["additional_usage_values"]
    assert additional_usage_values["cache_creation_input_tokens"] == 800


def test_sanitize_request_body_for_spend_logs_payload_basic():
    request_body = {
        "messages": [{"role": "user", "content": "Hello, how are you?"}],
    }
    assert _sanitize_request_body_for_spend_logs_payload(request_body) == request_body


def test_sanitize_request_body_for_spend_logs_payload_long_string():
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    # Create a string longer than MAX_STRING_LENGTH_PROMPT_IN_DB (2048)
    long_string = (
        "a" * 3000
    )  # Create a string longer than MAX_STRING_LENGTH_PROMPT_IN_DB
    request_body = {"text": long_string, "normal_text": "short text"}
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    # Calculate expected lengths: 35% start + 65% end + truncation message
    start_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.35)
    end_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.65)
    total_keep = start_chars + end_chars
    if total_keep > MAX_STRING_LENGTH_PROMPT_IN_DB:
        end_chars = MAX_STRING_LENGTH_PROMPT_IN_DB - start_chars

    skipped_chars = len(long_string) - (start_chars + end_chars)
    expected_truncation_message = f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. {LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE}) ..."
    expected_length = start_chars + len(expected_truncation_message) + end_chars

    assert len(sanitized["text"]) == expected_length
    assert sanitized["text"].startswith("a" * start_chars)
    assert sanitized["text"].endswith("a" * end_chars)
    assert expected_truncation_message in sanitized["text"]
    assert sanitized["normal_text"] == "short text"


def test_sanitize_request_body_for_spend_logs_payload_nested_dict():
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    # Create a string longer than MAX_STRING_LENGTH_PROMPT_IN_DB
    long_string = "a" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    request_body = {"outer": {"inner": {"text": long_string, "normal": "short"}}}
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    # Calculate expected lengths based on actual MAX_STRING_LENGTH_PROMPT_IN_DB
    start_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.35)
    end_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.65)
    total_keep = start_chars + end_chars
    if total_keep > MAX_STRING_LENGTH_PROMPT_IN_DB:
        end_chars = MAX_STRING_LENGTH_PROMPT_IN_DB - start_chars

    skipped_chars = len(long_string) - total_keep
    expected_truncation_message = f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. {LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE}) ..."
    expected_length = start_chars + len(expected_truncation_message) + end_chars

    assert len(sanitized["outer"]["inner"]["text"]) == expected_length
    assert sanitized["outer"]["inner"]["normal"] == "short"


def test_sanitize_request_body_for_spend_logs_payload_nested_list():
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    # Create a string longer than MAX_STRING_LENGTH_PROMPT_IN_DB
    long_string = "a" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    request_body = {
        "items": [{"text": long_string}, {"text": "short"}, [{"text": long_string}]]
    }
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    # Calculate expected lengths based on actual MAX_STRING_LENGTH_PROMPT_IN_DB
    start_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.35)
    end_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.65)
    total_keep = start_chars + end_chars
    if total_keep > MAX_STRING_LENGTH_PROMPT_IN_DB:
        end_chars = MAX_STRING_LENGTH_PROMPT_IN_DB - start_chars

    skipped_chars = len(long_string) - total_keep
    expected_truncation_message = f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. {LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE}) ..."
    expected_length = start_chars + len(expected_truncation_message) + end_chars

    assert len(sanitized["items"][0]["text"]) == expected_length
    assert sanitized["items"][1]["text"] == "short"
    assert len(sanitized["items"][2][0]["text"]) == expected_length


def test_sanitize_request_body_for_spend_logs_payload_non_string_values():
    request_body = {"number": 42, "boolean": True, "none": None, "float": 3.14}
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)
    assert sanitized == request_body


def test_sanitize_request_body_for_spend_logs_payload_empty():
    request_body: dict[str, Any] = {}
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)
    assert sanitized == request_body


def test_sanitize_request_body_for_spend_logs_payload_mixed_types():
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    # Create a string longer than MAX_STRING_LENGTH_PROMPT_IN_DB
    long_string = "a" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    request_body = {
        "text": long_string,
        "number": 42,
        "nested": {"list": ["short", long_string], "dict": {"key": long_string}},
    }
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    # Calculate expected lengths based on actual MAX_STRING_LENGTH_PROMPT_IN_DB
    start_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.35)
    end_chars = int(MAX_STRING_LENGTH_PROMPT_IN_DB * 0.65)
    total_keep = start_chars + end_chars
    if total_keep > MAX_STRING_LENGTH_PROMPT_IN_DB:
        end_chars = MAX_STRING_LENGTH_PROMPT_IN_DB - start_chars

    skipped_chars = len(long_string) - total_keep
    expected_truncation_message = f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. {LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE}) ..."
    expected_length = start_chars + len(expected_truncation_message) + end_chars

    assert len(sanitized["text"]) == expected_length
    assert sanitized["number"] == 42
    assert sanitized["nested"]["list"][0] == "short"
    assert len(sanitized["nested"]["list"][1]) == expected_length
    assert len(sanitized["nested"]["dict"]["key"]) == expected_length


def test_sanitize_request_body_for_spend_logs_payload_uses_runtime_env_override(
    monkeypatch: pytest.MonkeyPatch,
):
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    override_max = max(MAX_STRING_LENGTH_PROMPT_IN_DB + 1000, 6000)
    test_string = "a" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)

    # Simulate config-loaded env var being set after module import.
    monkeypatch.setenv("MAX_STRING_LENGTH_PROMPT_IN_DB", str(override_max))

    sanitized = _sanitize_request_body_for_spend_logs_payload({"text": test_string})
    assert sanitized["text"] == test_string


def test_sanitize_request_body_for_spend_logs_payload_circular_reference():
    # Create a circular reference
    a: dict[str, Any] = {}
    b: dict[str, Any] = {"a": a}
    a["b"] = b

    # Test that it handles circular reference without infinite recursion
    sanitized = _sanitize_request_body_for_spend_logs_payload(a)
    assert sanitized == {
        "b": {"a": {}}
    }  # Should return empty dict for circular reference


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_vector_store_request_for_spend_logs_payload_store_prompts_true(
    mock_should_store,
):
    # When _should_store_prompts_and_responses_in_spend_logs returns True
    mock_should_store.return_value = True

    # Sample vector store request metadata
    vector_store_request = [
        {
            "vector_store_search_response": {
                "data": [
                    {"content": [{"text": "sensitive information", "type": "text"}]}
                ]
            }
        }
    ]

    # When store_prompts is True, the original data should be returned unchanged
    result = _get_vector_store_request_for_spend_logs_payload(vector_store_request)
    assert result == vector_store_request
    assert (
        result[0]["vector_store_search_response"]["data"][0]["content"][0]["text"]
        == "sensitive information"
    )


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_vector_store_request_for_spend_logs_payload_store_prompts_false(
    mock_should_store,
):
    # When _should_store_prompts_and_responses_in_spend_logs returns False
    mock_should_store.return_value = False

    # Sample vector store request metadata
    vector_store_request = [
        {
            "vector_store_search_response": {
                "data": [
                    {"content": [{"text": "sensitive information", "type": "text"}]}
                ]
            }
        }
    ]

    # When store_prompts is False, text should be redacted
    result = _get_vector_store_request_for_spend_logs_payload(vector_store_request)
    assert result is not None
    assert (
        result[0]["vector_store_search_response"]["data"][0]["content"][0]["text"]
        == REDACTED_BY_LITELM_STRING
    )
    # Ensure other fields are unchanged
    assert (
        result[0]["vector_store_search_response"]["data"][0]["content"][0]["type"]
        == "text"
    )


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_vector_store_request_for_spend_logs_payload_null_input(mock_should_store):
    # When input is None
    mock_should_store.return_value = False
    result = _get_vector_store_request_for_spend_logs_payload(None)
    assert result is None


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_messages_for_spend_logs_realtime_returns_messages(mock_should_store):
    """
    Test that _get_messages_for_spend_logs_payload returns messages
    for realtime calls when store_prompts_in_spend_logs is True.
    """
    mock_should_store.return_value = True
    realtime_messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the weather today?"},
    ]
    payload = cast(
        StandardLoggingPayload,
        {
            "call_type": "_arealtime",
            "messages": realtime_messages,
        },
    )
    result = _get_messages_for_spend_logs_payload(payload)
    parsed = json.loads(result)
    assert len(parsed) == 2
    assert parsed[0]["role"] == "system"
    assert parsed[0]["content"] == "You are a helpful assistant."
    assert parsed[1]["role"] == "user"
    assert parsed[1]["content"] == "What is the weather today?"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_messages_for_spend_logs_strips_null_bytes(mock_should_store):
    """Regression for PostgreSQL 22P05: NUL bytes must be stripped from messages."""
    mock_should_store.return_value = True
    payload = cast(
        StandardLoggingPayload,
        {
            "call_type": "_arealtime",
            "messages": [{"role": "user", "content": "hello\x00world"}],
        },
    )
    result = _get_messages_for_spend_logs_payload(payload)
    assert "\\u0000" not in result
    parsed = json.loads(result)
    assert parsed[0]["content"] == "helloworld"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_messages_for_spend_logs_realtime_empty_when_disabled(mock_should_store):
    """
    Test that _get_messages_for_spend_logs_payload returns '{}' for realtime calls
    when store_prompts_in_spend_logs is False.
    """
    mock_should_store.return_value = False
    payload = cast(
        StandardLoggingPayload,
        {
            "call_type": "_arealtime",
            "messages": [{"role": "user", "content": "Hello"}],
        },
    )
    result = _get_messages_for_spend_logs_payload(payload)
    assert result == "{}"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_messages_for_spend_logs_non_realtime_returns_empty(mock_should_store):
    """
    Test that _get_messages_for_spend_logs_payload returns '{}' for non-realtime
    calls even when store_prompts_in_spend_logs is True.
    """
    mock_should_store.return_value = True
    payload = cast(
        StandardLoggingPayload,
        {
            "call_type": "acompletion",
            "messages": [{"role": "user", "content": "Hello"}],
        },
    )
    result = _get_messages_for_spend_logs_payload(payload)
    assert result == "{}"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_response_for_spend_logs_payload_truncates_large_base64(mock_should_store):
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    mock_should_store.return_value = True
    large_text = "A" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    payload = cast(
        StandardLoggingPayload,
        {
            "response": {
                "data": [
                    {
                        "b64_json": large_text,
                        "other_field": "value",
                    }
                ]
            }
        },
    )

    response_json = _get_response_for_spend_logs_payload(payload)
    parsed = json.loads(response_json)
    truncated_value = parsed["data"][0]["b64_json"]
    assert len(truncated_value) < len(large_text)
    assert LITELLM_TRUNCATED_PAYLOAD_FIELD in truncated_value
    assert parsed["data"][0]["other_field"] == "value"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_response_for_spend_logs_payload_strips_null_bytes(mock_should_store):
    """Regression for PostgreSQL 22P05: NUL bytes must be stripped from response."""
    mock_should_store.return_value = True
    payload = cast(
        StandardLoggingPayload,
        {"response": {"content": "answer\x00here"}},
    )
    response_json = _get_response_for_spend_logs_payload(payload)
    assert "\\u0000" not in response_json
    assert json.loads(response_json)["content"] == "answerhere"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_get_response_for_spend_logs_payload_truncates_large_embedding(
    mock_should_store,
):
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    mock_should_store.return_value = True
    embedding_values = [
        round(i * 0.0001, 6) for i in range(MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    ]
    large_embedding = json.dumps(embedding_values)
    payload = cast(
        StandardLoggingPayload,
        {
            "response": {
                "data": [
                    {
                        "embedding": large_embedding,
                        "other_field": "value",
                    }
                ]
            }
        },
    )

    response_json = _get_response_for_spend_logs_payload(payload)
    parsed = json.loads(response_json)
    truncated_value = parsed["data"][0]["embedding"]

    assert isinstance(truncated_value, str)
    assert len(truncated_value) < len(large_embedding)
    assert LITELLM_TRUNCATED_PAYLOAD_FIELD in truncated_value
    assert parsed["data"][0]["other_field"] == "value"


def test_truncation_includes_db_safeguard_note():
    """
    Test that truncated content includes the DB safeguard note explaining
    that full data is available in OTEL/other logging integrations.
    """
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    large_error = "Error: " + "x" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 1000)
    request_body = {"error_trace": large_error}
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    truncated = sanitized["error_trace"]
    assert LITELLM_TRUNCATED_PAYLOAD_FIELD in truncated
    assert LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE in truncated
    assert "DB storage safeguard" in truncated
    assert (
        "logging callbacks" in truncated.lower()
        or "logging integrations" in truncated.lower()
        or "logging callbacks" in truncated
    )


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_response_truncation_logs_info_message(mock_should_store):
    """
    Test that when response is truncated before DB storage, an info log is emitted
    noting that full data is available in OTEL/other integrations.
    """
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    mock_should_store.return_value = True
    large_text = "B" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    payload = cast(
        StandardLoggingPayload,
        {"response": {"data": [{"content": large_text}]}},
    )

    with patch(
        "litellm.proxy.spend_tracking.spend_tracking_utils.verbose_proxy_logger"
    ) as mock_logger:
        _get_response_for_spend_logs_payload(payload)
        mock_logger.info.assert_called_once()
        log_msg = mock_logger.info.call_args[0][0]
        assert "response was truncated" in log_msg


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_request_body_truncation_logs_info_message(mock_should_store):
    """
    Test that when request body is truncated before DB storage, an info log is emitted.
    """
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    mock_should_store.return_value = True
    large_prompt = "C" * (MAX_STRING_LENGTH_PROMPT_IN_DB + 500)
    litellm_params = {
        "proxy_server_request": {
            "body": {"messages": [{"role": "user", "content": large_prompt}]}
        }
    }

    with patch(
        "litellm.proxy.spend_tracking.spend_tracking_utils.verbose_proxy_logger"
    ) as mock_logger:
        _get_proxy_server_request_for_spend_logs_payload(
            metadata={}, litellm_params=litellm_params, kwargs={}
        )
        mock_logger.info.assert_called_once()
        log_msg = mock_logger.info.call_args[0][0]
        assert "request body was truncated" in log_msg


def test_safe_dumps_handles_circular_references():
    """Test that safe_dumps can handle circular references without raising exceptions"""

    # Create a circular reference
    obj1 = {"name": "obj1"}
    obj2 = {"name": "obj2", "ref": obj1}
    obj1["ref"] = obj2  # This creates a circular reference

    # This should not raise an exception
    result = safe_dumps(obj1)

    # Should be a valid JSON string
    assert isinstance(result, str)

    # Should contain placeholder for circular reference
    assert "CircularReference Detected" in result

    # Should be parseable as JSON
    parsed = json.loads(result)
    assert parsed["name"] == "obj1"
    assert parsed["ref"]["name"] == "obj2"


def test_safe_dumps_normal_objects():
    """Test that safe_dumps works correctly with normal objects"""

    normal_obj = {
        "string": "test",
        "number": 42,
        "boolean": True,
        "null": None,
        "list": [1, 2, 3],
        "nested": {"key": "value"},
    }

    result = safe_dumps(normal_obj)

    # Should be a valid JSON string that can be parsed
    assert isinstance(result, str)
    parsed = json.loads(result)
    assert parsed == normal_obj


def test_safe_dumps_complex_metadata_like_object():
    """Test with a complex metadata-like object similar to what caused the issue"""

    # Simulate a complex metadata object
    metadata = {
        "user_api_key": "test-key",
        "model": "gpt-4",
        "usage": {"total_tokens": 100},
        "mcp_tool_call_metadata": {
            "name": "test_tool",
            "arguments": {"param": "value"},
        },
    }

    # Add a potential circular reference
    usage_detail = {"parent_metadata": metadata}
    metadata["usage"]["detail"] = usage_detail

    # This should not raise an exception
    result = safe_dumps(metadata)

    # Should be a valid JSON string
    assert isinstance(result, str)

    # Should be parseable as JSON
    parsed = json.loads(result)
    assert parsed["user_api_key"] == "test-key"
    assert parsed["model"] == "gpt-4"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_api_key_preserved_when_standard_logging_payload_is_none():
    """
    Critical - Product incident was caused by this bug.

    Test that api_key is NOT set to empty string when standard_logging_payload is None.

    This is a regression test for a bug where:
    - On failed requests (bad request errors), standard_logging_payload is None
    - The else block was incorrectly setting api_key = ""
    - This caused empty api_key in DailyUserSpend table despite SpendLogs having the correct key

    Expected behavior:
    - api_key from metadata should be extracted and hashed
    - Even when standard_logging_payload is None, the api_key should be preserved
    - The returned payload should have the hashed api_key, not empty string
    """
    # Setup: Simulate a failed request scenario
    test_api_key = "sk-WLi4iRn4JmbVlTaYw12IOA"

    # Create kwargs similar to what's passed during a bad request error
    kwargs = {
        "model": "openai/gpt-4.1",
        "messages": [{"role": "user", "content": "Hello"}],
        "call_type": "acompletion",
        "litellm_params": {
            "metadata": {
                "user_api_key": test_api_key,  # This is the key that should be preserved
                "user_api_key_user_id": "test_user",
                "user_api_key_team_id": "test_team",
            }
        },
        # Note: No 'standard_logging_object' in kwargs - simulating failure case
    }

    # Create a mock error response (bad request)
    response_obj = Exception("BadRequestError: Invalid parameter 'usersss'")

    # Create timestamps
    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    # Call get_logging_payload
    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    # CRITICAL ASSERTION: api_key should NOT be empty string
    assert payload["api_key"] != "", (
        "BUG: api_key is empty! When standard_logging_payload is None, "
        "the api_key from metadata should be preserved and hashed."
    )

    # The api_key should be hashed (not the raw key)
    assert (
        payload["api_key"] != test_api_key
    ), "api_key should be hashed, not the raw key"

    # The api_key should be a valid hash (64 character hex string for SHA256)
    assert (
        len(payload["api_key"]) == 64
    ), f"Expected 64 character hash, got {len(payload['api_key'])} characters"

    # Verify other fields are set correctly
    assert payload["model"] == "openai/gpt-4.1"
    assert payload["user"] == "test_user"


@pytest.mark.asyncio
@patch("litellm.proxy.proxy_server.master_key", "sk-master-key")
@patch("litellm.proxy.proxy_server.general_settings", {})
async def test_api_key_preserved_through_failure_hook_to_database():
    """
    CRITICAL E2E TEST: Validates the COMPLETE code path from failure hook to database.

    This is THE comprehensive test that protects against the production incident.
    It tests the EXACT flow that caused the bug:

    1. async_post_call_failure_hook is called with api_key in UserAPIKeyAuth
    2. Failure hook calls update_database with the token parameter
    3. update_database calls get_logging_payload to create payload
    4. BUG WAS HERE: get_logging_payload set api_key = "" when standard_logging_payload was None
    5. Empty api_key was written to DailyUserSpend table

    This test validates the ENTIRE flow to ensure the bug cannot regress.
    If this test fails in CI/CD, the build MUST fail.
    """
    from litellm.proxy._types import UserAPIKeyAuth
    from litellm.proxy.hooks.proxy_track_cost_callback import _ProxyDBLogger
    from litellm.proxy.utils import hash_token

    # Setup
    test_api_key = "sk-test-critical-e2e-key"
    hashed_key = hash_token(test_api_key)

    # Track what payload gets created
    captured_payloads = []

    async def mock_update_database(
        token,
        response_cost,
        user_id,
        end_user_id,
        team_id,
        kwargs,
        completion_response,
        start_time,
        end_time,
        org_id,
    ):
        """Mock update_database and capture the payload it creates"""
        from litellm.proxy.spend_tracking.spend_tracking_utils import (
            get_logging_payload,
        )

        # Call get_logging_payload EXACTLY as update_database does
        payload = get_logging_payload(
            kwargs=kwargs,
            response_obj=completion_response,
            start_time=start_time,
            end_time=end_time,
        )

        captured_payloads.append(
            {
                "token": token,
                "payload": payload,
            }
        )

    # Mock dependencies
    mock_db_writer = MagicMock()
    mock_db_writer.update_database = AsyncMock(side_effect=mock_update_database)

    mock_proxy_logging_obj = MagicMock()
    mock_proxy_logging_obj.db_spend_update_writer = mock_db_writer

    # Create UserAPIKeyAuth (what the failure hook receives)
    user_api_key_dict = UserAPIKeyAuth(
        api_key=hashed_key,
        user_id="test_user",
        team_id="test_team",
        max_budget=None,
        spend=0.0,
        key_alias="test-key",
        budget_reset_at=None,
        user_email=None,
        org_id="test_org",
        team_alias=None,
        end_user_id=None,
        request_route="/chat/completions",
        metadata={},
    )

    # Request data with bad parameter (triggers failure)
    request_data = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": "test"}],
        "invalid_param": "causes_400_error",  # BAD PARAMETER
        "litellm_params": {
            "metadata": {
                "user_api_key": hashed_key,
                "user_api_key_user_id": "test_user",
                "user_api_key_team_id": "test_team",
            }
        },
    }

    exception = Exception("BadRequestError: Invalid parameter 'invalid_param'")

    # Execute the ACTUAL failure hook code path
    logger = _ProxyDBLogger()

    with patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging_obj):
        await logger.async_post_call_failure_hook(
            request_data=request_data,
            original_exception=exception,
            user_api_key_dict=user_api_key_dict,
            traceback_str=None,
        )

        await asyncio.sleep(0.1)  # Wait for async operations

    # =========================================================================
    # CRITICAL ASSERTIONS - If ANY fail, the production bug has regressed!
    # =========================================================================

    assert len(captured_payloads) == 1, "update_database should be called once"

    data = captured_payloads[0]
    payload = data["payload"]
    payload_api_key = payload.get("api_key")

    # THE CRITICAL ASSERTION - This would fail with the original bug!
    assert payload_api_key != "", (
        "🚨 CRITICAL BUG: payload['api_key'] is empty! "
        "This is the EXACT production incident bug. "
        "get_logging_payload() is setting api_key = '' when "
        "standard_logging_payload is None (failure case)."
    )

    assert payload_api_key is not None, "🚨 CRITICAL: payload['api_key'] is None!"

    assert (
        payload_api_key == hashed_key
    ), f"🚨 CRITICAL: Expected api_key={hashed_key}, got {payload_api_key}"

    # Verify token parameter matches
    assert data["token"] == hashed_key, f"Token parameter should be {hashed_key}"

    # Verify other fields
    assert payload.get("model") == "gpt-3.5-turbo"
    assert payload.get("user") == "test_user"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_includes_agent_id_from_kwargs():
    """
    Test that get_logging_payload extracts agent_id from kwargs and includes it in the payload.
    """
    test_agent_id = "agent-uuid-12345"

    kwargs = {
        "model": "a2a_agent/test-agent",
        "custom_llm_provider": "a2a_agent",
        "agent_id": test_agent_id,
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
            }
        },
    }

    response_obj = {
        "id": "test-response-123",
        "jsonrpc": "2.0",
        "result": {"status": "completed"},
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    assert (
        payload["agent_id"] == test_agent_id
    ), f"Expected agent_id '{test_agent_id}', got '{payload.get('agent_id')}'"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_includes_overhead_in_spend_logs_metadata():
    """
    Test that get_logging_payload extracts litellm_overhead_time_ms from hidden_params
    and stores it in spend_logs_metadata within the metadata JSON.
    """
    test_overhead_ms = 123.45

    # Create StandardLoggingPayload with hidden_params containing overhead
    standard_logging_payload = StandardLoggingPayload(
        id="test-id-123",
        call_type="completion",
        stream=False,
        response_cost=0.001,
        status="success",
        total_tokens=100,
        prompt_tokens=50,
        completion_tokens=50,
        startTime=1234567890.0,
        endTime=1234567891.0,
        completionStartTime=None,
        model_map_information=StandardLoggingModelInformation(
            model_map_key="gpt-3.5-turbo", model_map_value=None
        ),
        model="gpt-3.5-turbo",
        model_id="model-123",
        model_group="openai",
        custom_llm_provider="openai",
        api_base="https://api.openai.com",
        metadata=StandardLoggingMetadata(
            user_api_key_hash="test_hash",
            user_api_key_alias=None,
            user_api_key_team_id=None,
            user_api_key_org_id=None,
            user_api_key_user_id=None,
            user_api_key_team_alias=None,
            spend_logs_metadata=None,
            requester_ip_address=None,
            requester_metadata=None,
            user_api_key_end_user_id=None,
        ),
        cache_hit=False,
        cache_key=None,
        saved_cache_cost=0.0,
        request_tags=[],
        end_user=None,
        requester_ip_address=None,
        messages=[],
        response={},
        error_str=None,
        model_parameters={},
        hidden_params=StandardLoggingHiddenParams(
            model_id="model-123",
            cache_key=None,
            api_base="https://api.openai.com",
            response_cost="0.001",
            litellm_overhead_time_ms=test_overhead_ms,
            additional_headers=None,
            batch_models=None,
            litellm_model_name=None,
            usage_object=None,
        ),
    )

    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
            }
        },
        "standard_logging_object": standard_logging_payload,
    }

    response_obj = {
        "id": "test-response-123",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    # Parse the metadata JSON string
    metadata_json = payload.get("metadata")
    assert metadata_json is not None, "metadata should not be None"

    metadata = json.loads(metadata_json)

    # Verify overhead is stored directly in metadata
    assert (
        metadata.get("litellm_overhead_time_ms") == test_overhead_ms
    ), f"Expected overhead '{test_overhead_ms}', got '{metadata.get('litellm_overhead_time_ms')}'"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_strips_null_bytes_from_request_tags():
    """Regression for PostgreSQL 22P05: NUL bytes must be stripped from request_tags."""
    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
                "tags": ["clean-tag", "bad\x00tag"],
            }
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj={},
        start_time=start_time,
        end_time=end_time,
    )

    request_tags = payload.get("request_tags")
    assert request_tags is not None
    assert "\\u0000" not in request_tags
    assert json.loads(request_tags) == ["clean-tag", "badtag"]


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_handles_missing_overhead_gracefully():
    """
    Test that get_logging_payload handles missing overhead gracefully
    (backward compatibility - when overhead is not present, it should not break).
    """
    # Create StandardLoggingPayload WITHOUT overhead in hidden_params
    standard_logging_payload = StandardLoggingPayload(
        id="test-id-456",
        call_type="completion",
        stream=False,
        response_cost=0.001,
        status="success",
        total_tokens=100,
        prompt_tokens=50,
        completion_tokens=50,
        startTime=1234567890.0,
        endTime=1234567891.0,
        completionStartTime=None,
        model_map_information=StandardLoggingModelInformation(
            model_map_key="gpt-3.5-turbo", model_map_value=None
        ),
        model="gpt-3.5-turbo",
        model_id="model-123",
        model_group="openai",
        custom_llm_provider="openai",
        api_base="https://api.openai.com",
        metadata=StandardLoggingMetadata(
            user_api_key_hash="test_hash",
            user_api_key_alias=None,
            user_api_key_team_id=None,
            user_api_key_org_id=None,
            user_api_key_user_id=None,
            user_api_key_team_alias=None,
            spend_logs_metadata=None,
            requester_ip_address=None,
            requester_metadata=None,
            user_api_key_end_user_id=None,
        ),
        cache_hit=False,
        cache_key=None,
        saved_cache_cost=0.0,
        request_tags=[],
        end_user=None,
        requester_ip_address=None,
        messages=[],
        response={},
        error_str=None,
        model_parameters={},
        hidden_params=StandardLoggingHiddenParams(
            model_id="model-123",
            cache_key=None,
            api_base="https://api.openai.com",
            response_cost="0.001",
            litellm_overhead_time_ms=None,  # No overhead
            additional_headers=None,
            batch_models=None,
            litellm_model_name=None,
            usage_object=None,
        ),
    )

    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
            }
        },
        "standard_logging_object": standard_logging_payload,
    }

    response_obj = {
        "id": "test-response-456",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    # Should not raise an exception
    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    # Parse the metadata JSON string
    metadata_json = payload.get("metadata")
    assert metadata_json is not None, "metadata should not be None"

    metadata = json.loads(metadata_json)

    # When overhead is None, litellm_overhead_time_ms should be None or not present
    assert (
        metadata.get("litellm_overhead_time_ms") is None
    ), "litellm_overhead_time_ms should be None when overhead is not provided"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_spend_logs_redacts_request_and_response_when_turn_off_message_logging_enabled(
    mock_should_store,
):
    """
    Test that both request body and response are redacted when turn_off_message_logging is enabled.
    """
    mock_should_store.return_value = True

    # Test request redaction
    litellm_params = {
        "proxy_server_request": {
            "body": {
                "messages": [{"role": "user", "content": "secret message"}],
                "model": "gpt-4",
            }
        }
    }
    metadata = {}
    kwargs = {
        "litellm_params": litellm_params,
        "standard_callback_dynamic_params": {
            "turn_off_message_logging": True,
        },
    }

    request_result = _get_proxy_server_request_for_spend_logs_payload(
        metadata=metadata, litellm_params=litellm_params, kwargs=kwargs
    )

    parsed_request = json.loads(request_result)
    assert parsed_request["messages"] == [
        {"role": "user", "content": "redacted-by-litellm"}
    ]
    assert parsed_request["model"] == "gpt-4"

    # Test response redaction - use dict response to verify redaction
    response_dict = {
        "id": "test-id",
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": "secret response"},
            }
        ],
        "model": "gpt-4",
    }
    payload = cast(
        StandardLoggingPayload,
        {"response": response_dict},
    )

    response_result = _get_response_for_spend_logs_payload(
        payload=payload, kwargs=kwargs
    )

    # When redaction is enabled and response is a dict (not ModelResponse),
    # perform_redaction redacts content in-place within the choices structure
    parsed_response = json.loads(response_result)
    assert parsed_response["choices"][0]["message"]["content"] == "redacted-by-litellm"
    assert parsed_response["choices"][0]["message"]["role"] == "assistant"


@patch("litellm.secret_managers.main.get_secret_bool")
def test_should_store_prompts_and_responses_in_spend_logs_case_insensitive_string(
    mock_get_secret_bool,
):
    """
    Test that _should_store_prompts_and_responses_in_spend_logs handles
    case-insensitive string values for store_prompts_in_spend_logs in general_settings.
    """
    # Test case-insensitive string "true" variations
    for true_value in ["true", "TRUE", "True", "TrUe"]:
        with patch(
            "litellm.proxy.proxy_server.general_settings",
            {"store_prompts_in_spend_logs": true_value},
        ):
            mock_get_secret_bool.return_value = False  # Ensure env var is False
            result = _should_store_prompts_and_responses_in_spend_logs()
            assert result is True, f"Expected True for '{true_value}', got {result}"

    # Test boolean True
    with patch(
        "litellm.proxy.proxy_server.general_settings",
        {"store_prompts_in_spend_logs": True},
    ):
        mock_get_secret_bool.return_value = False
        result = _should_store_prompts_and_responses_in_spend_logs()
        assert result is True, f"Expected True for boolean True, got {result}"

    # Test that non-true values fall back to environment variable
    for false_value in [False, None, "false", "FALSE", "False", "anything"]:
        with patch(
            "litellm.proxy.proxy_server.general_settings",
            {"store_prompts_in_spend_logs": false_value},
        ):
            # When env var is True, should return True
            mock_get_secret_bool.return_value = True
            result = _should_store_prompts_and_responses_in_spend_logs()
            assert (
                result is True
            ), f"Expected True (from env var) for '{false_value}', got {result}"

            # When env var is False, should return False
            mock_get_secret_bool.return_value = False
            result = _should_store_prompts_and_responses_in_spend_logs()
            assert (
                result is False
            ), f"Expected False (from env var) for '{false_value}', got {result}"

    # Test when general_settings doesn't have the key at all
    with patch("litellm.proxy.proxy_server.general_settings", {}):
        mock_get_secret_bool.return_value = True
        result = _should_store_prompts_and_responses_in_spend_logs()
        assert (
            result is True
        ), "Expected True (from env var) when key missing, got False"

        mock_get_secret_bool.return_value = False
        result = _should_store_prompts_and_responses_in_spend_logs()
        assert (
            result is False
        ), "Expected False (from env var) when key missing, got True"


def test_get_spend_logs_metadata_guardrail_info_fallback_from_metadata():
    """
    When standard_logging_payload is None (e.g. guardrail blocks before LLM call),
    guardrail_information should fall back to reading from metadata's
    standard_logging_guardrail_information field.
    """
    guardrail_info = [
        {
            "guardrail_name": "content_filter",
            "guardrail_provider": "litellm",
            "guardrail_mode": "pre_call",
            "guardrail_status": "guardrail_intervened",
            "guardrail_response": "Content blocked",
        }
    ]
    metadata = {
        "user_api_key": "test-key",
        "standard_logging_guardrail_information": guardrail_info,
    }

    result = _get_spend_logs_metadata(
        metadata=metadata,
        guardrail_information=None,
    )
    # When guardrail_information param is None, should NOT fall back
    # (the caller is responsible for passing it)
    assert result["guardrail_information"] is None


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_redacts_all_prompt_carrying_fields_when_flag_false(
    mock_should_store,
):
    """
    match_details and classification are declared as structured metadata but
    in-tree writers (litellm_content_filter, block_code_execution) inline
    raw prompt content into them, so they leak the same way
    guardrail_request/guardrail_response do. Redaction must cover all four.
    """
    mock_should_store.return_value = False
    guardrail_info = [
        {
            "guardrail_name": "demo-echo-guard",
            "guardrail_status": "success",
            "guardrail_request": {"messages": [{"role": "user", "content": "hi"}]},
            "guardrail_response": {"evaluated_input": "hi"},
            "match_details": [{"type": "pattern", "snippet": "hi", "action_taken": "log"}],
            "classification": {"intent": "x", "evidence": [{"match": "hi"}]},
            "guardrail_action": "NONE",
        }
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result is not None
    entry = result[0]
    assert entry["guardrail_request"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert entry["match_details"] == REDACTED_BY_LITELM_STRING
    assert entry["classification"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_name"] == "demo-echo-guard"
    assert entry["guardrail_status"] == "success"
    assert entry["guardrail_action"] == "NONE"


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_redacts_prompt_fields_when_flag_false(
    mock_should_store,
):
    """
    LIT-4314 Issue A regression: with store_prompts_in_spend_logs=False,
    guardrail_request and guardrail_response must be redacted before they
    land in LiteLLM_SpendLogs.metadata, while every other field on the
    entry is preserved bit-for-bit.
    """
    mock_should_store.return_value = False
    guardrail_info = [
        {
            "guardrail_name": "demo-echo-guard",
            "guardrail_provider": "custom",
            "guardrail_mode": "pre_call",
            "guardrail_status": "success",
            "guardrail_request": {
                "messages": [{"role": "user", "content": "Say hi in 3 words"}],
            },
            "guardrail_response": {
                "evaluated_input": "Say hi in 3 words",
                "verdict": "allow",
            },
            "start_time": 1_700_000_000.0,
            "end_time": 1_700_000_000.5,
            "duration": 0.5,
            "guardrail_id": "gd-42",
            "masked_entity_count": {"EMAIL": 1},
            "violation_categories": ["prompt_injection"],
            "risk_score": 3.5,
            "guardrail_action": "NONE",
        }
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result is not None
    assert len(result) == 1
    entry = result[0]
    assert entry["guardrail_request"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_name"] == "demo-echo-guard"
    assert entry["guardrail_provider"] == "custom"
    assert entry["guardrail_mode"] == "pre_call"
    assert entry["guardrail_status"] == "success"
    assert entry["start_time"] == 1_700_000_000.0
    assert entry["end_time"] == 1_700_000_000.5
    assert entry["duration"] == 0.5
    assert entry["guardrail_id"] == "gd-42"
    assert entry["masked_entity_count"] == {"EMAIL": 1}
    assert entry["violation_categories"] == ["prompt_injection"]
    assert entry["risk_score"] == 3.5
    assert entry["guardrail_action"] == "NONE"

    assert guardrail_info[0]["guardrail_request"] == {
        "messages": [{"role": "user", "content": "Say hi in 3 words"}],
    }
    assert guardrail_info[0]["guardrail_response"] == {
        "evaluated_input": "Say hi in 3 words",
        "verdict": "allow",
    }


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_preserves_guardrail_usage_when_flag_false(
    mock_should_store,
):
    """
    LIT-5650 regression: provider-reported billable usage counters live in
    guardrail_usage, a sibling of guardrail_response, precisely so the
    default spend-log redaction cannot drop them. The response blob (which
    also embeds a usage copy) must still be redacted wholesale.
    """
    mock_should_store.return_value = False
    guardrail_info = [
        {
            "guardrail_name": "bedrock-guard",
            "guardrail_status": "guardrail_intervened",
            "guardrail_response": {
                "action": "GUARDRAIL_INTERVENED",
                "outputs": [{"text": "Sorry, the model cannot answer this question."}],
                "usage": {"topicPolicyUnits": 1, "contentPolicyUnits": 1},
            },
            "guardrail_usage": {"topicPolicyUnits": 1, "contentPolicyUnits": 1, "wordPolicyUnits": 0},
        }
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result is not None
    entry = result[0]
    assert entry["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_usage"] == {"topicPolicyUnits": 1, "contentPolicyUnits": 1, "wordPolicyUnits": 0}


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_passthrough_when_flag_true(
    mock_should_store,
):
    """
    When store_prompts_in_spend_logs=True the sanitizer must be a no-op so
    operators who explicitly opted in still see full guardrail payloads.
    """
    mock_should_store.return_value = True
    guardrail_info = [
        {
            "guardrail_name": "content_filter",
            "guardrail_status": "success",
            "guardrail_request": {"messages": [{"role": "user", "content": "hi"}]},
            "guardrail_response": {"verdict": "allow"},
        }
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result == guardrail_info


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_none_passthrough(mock_should_store):
    mock_should_store.return_value = False
    assert _sanitize_guardrail_information_for_spend_logs(None) is None


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_normalizes_bare_dict_input(mock_should_store):
    """
    Regression: xecguard (xecguard.py:246) assigns a bare dict to
    standard_logging_object["guardrail_information"] even though the typed
    contract is Optional[List[...]]. Without defensive normalization here,
    the for-loop would iterate the dict's string keys and _redact...
    would TypeError on {**"guardrail_name"}, taking down the entire
    spend-log write via update_database's broad except.
    """
    mock_should_store.return_value = False
    bare_dict_entry = {
        "guardrail_name": "xecguard",
        "guardrail_status": "success",
        "guardrail_response": {"decision": "SAFE", "raw_prompt": "hi"},
        "start_time": 1.0,
        "end_time": 2.0,
        "duration": 1.0,
    }

    result = _sanitize_guardrail_information_for_spend_logs(bare_dict_entry)

    assert result is not None
    assert isinstance(result, list)
    assert len(result) == 1
    entry = result[0]
    assert entry["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_name"] == "xecguard"
    assert entry["guardrail_status"] == "success"
    assert entry["start_time"] == 1.0


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_drops_non_dict_items_in_list(mock_should_store):
    """
    A stray non-dict item in the list (e.g. from a buggy caller that
    accidentally appends a string) should be silently skipped instead of
    crashing the spend-log write.
    """
    mock_should_store.return_value = False
    mixed_input = [
        {"guardrail_name": "x", "guardrail_response": {"leak": "hi"}},
        "not-a-dict",
        None,
    ]

    result = _sanitize_guardrail_information_for_spend_logs(mixed_input)

    assert result == [{"guardrail_name": "x", "guardrail_response": REDACTED_BY_LITELM_STRING}]


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_preserves_absent_prompt_fields(mock_should_store):
    """
    Entries that never carried guardrail_request or guardrail_response must
    not gain those keys after sanitization; consumers keying on presence
    (`"guardrail_request" in entry`) would otherwise flip from absent to
    the sentinel string.
    """
    mock_should_store.return_value = False
    guardrail_info = [
        {
            "guardrail_name": "demo-echo-guard",
            "guardrail_status": "success",
            "guardrail_response": {"verdict": "allow", "evaluated_input": "hi"},
        }
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result is not None
    entry = result[0]
    assert "guardrail_request" not in entry
    assert entry["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert entry["guardrail_name"] == "demo-echo-guard"
    assert entry["guardrail_status"] == "success"


@patch("litellm.proxy.proxy_server.master_key", "sk-master")
@patch(
    "litellm.proxy.proxy_server.general_settings",
    {"store_prompts_in_spend_logs": False},
)
def test_get_logging_payload_redacts_guardrail_prompt_fields_when_flag_false():
    """
    End-to-end wire-in check: get_logging_payload -> _get_spend_logs_metadata
    -> sanitizer. Without the wire-in at line 139, the raw guardrail_response
    lands in payload["metadata"] verbatim.
    """
    guardrail_info = [
        {
            "guardrail_name": "demo-echo-guard",
            "guardrail_provider": "custom",
            "guardrail_status": "success",
            "guardrail_request": {"messages": [{"role": "user", "content": "secret"}]},
            "guardrail_response": {"evaluated_input": "secret"},
        }
    ]
    kwargs = {
        "model": "gpt-4o-mini",
        "litellm_call_id": "test-call-id",
        "litellm_params": {
            "metadata": {
                "user_api_key": "test-key",
                "standard_logging_guardrail_information": guardrail_info,
            },
            "proxy_server_request": {},
        },
    }

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj={},
        start_time=datetime.datetime.now(tz=timezone.utc),
        end_time=datetime.datetime.now(tz=timezone.utc),
    )

    metadata_result = json.loads(payload["metadata"])
    stored = metadata_result["guardrail_information"][0]
    assert stored["guardrail_request"] == REDACTED_BY_LITELM_STRING
    assert stored["guardrail_response"] == REDACTED_BY_LITELM_STRING
    assert stored["guardrail_name"] == "demo-echo-guard"
    assert stored["guardrail_status"] == "success"


def test_get_logging_payload_guardrail_info_when_no_standard_logging_payload():
    """
    When a guardrail blocks a request before the LLM call, the standard_logging_object
    is not set on request_data. In this case, get_logging_payload should still include
    guardrail_information from the metadata.

    This is the bug fix for: guardrail failures not showing GuardrailViewer in the UI.
    """
    guardrail_info = [
        {
            "guardrail_name": "content_filter",
            "guardrail_provider": "litellm",
            "guardrail_mode": "pre_call",
            "guardrail_status": "guardrail_intervened",
            "guardrail_response": "Content blocked",
        }
    ]
    # Simulate request_data as it looks when a guardrail blocks before LLM call
    kwargs = {
        "model": "gpt-4",
        "litellm_call_id": "test-call-id",
        "litellm_params": {
            "metadata": {
                "user_api_key": "test-key",
                "standard_logging_guardrail_information": guardrail_info,
            },
            "proxy_server_request": {},
        },
        # No "standard_logging_object" key - this is the failure case
    }

    with patch("litellm.proxy.proxy_server.master_key", "sk-master"):
        with patch(
            "litellm.proxy.proxy_server.general_settings",
            {"store_prompts_in_spend_logs": True},
        ):
            payload = get_logging_payload(
                kwargs=kwargs,
                response_obj={},
                start_time=datetime.datetime.now(tz=timezone.utc),
                end_time=datetime.datetime.now(tz=timezone.utc),
            )

    metadata_result = json.loads(payload["metadata"])
    assert metadata_result["guardrail_information"] == guardrail_info


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_includes_retry_info_in_spend_logs_metadata():
    """
    Test that retry info (attempted_retries, max_retries) from metadata
    is included in the spend logs metadata JSON.
    """
    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
                "attempted_retries": 2,
                "max_retries": 3,
            }
        },
        "standard_logging_object": StandardLoggingPayload(
            id="test-retry-123",
            call_type="completion",
            stream=False,
            response_cost=0.001,
            status="success",
            total_tokens=100,
            prompt_tokens=50,
            completion_tokens=50,
            startTime=1234567890.0,
            endTime=1234567891.0,
            completionStartTime=None,
            model_map_information=StandardLoggingModelInformation(
                model_map_key="gpt-3.5-turbo", model_map_value=None
            ),
            model="gpt-3.5-turbo",
            model_id="model-123",
            model_group="openai",
            custom_llm_provider="openai",
            api_base="https://api.openai.com",
            metadata=StandardLoggingMetadata(
                user_api_key_hash="test_hash",
                user_api_key_alias=None,
                user_api_key_team_id=None,
                user_api_key_org_id=None,
                user_api_key_user_id=None,
                user_api_key_team_alias=None,
                spend_logs_metadata=None,
                requester_ip_address=None,
                requester_metadata=None,
                user_api_key_end_user_id=None,
            ),
            cache_hit=False,
            cache_key=None,
            saved_cache_cost=0.0,
            request_tags=[],
            end_user=None,
            requester_ip_address=None,
            messages=[],
            response={},
            error_str=None,
            model_parameters={},
            hidden_params=StandardLoggingHiddenParams(
                model_id="model-123",
                cache_key=None,
                api_base="https://api.openai.com",
                response_cost="0.001",
                litellm_overhead_time_ms=None,
                additional_headers=None,
                batch_models=None,
                litellm_model_name=None,
                usage_object=None,
            ),
        ),
    }

    response_obj = {
        "id": "test-response-retry",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    metadata = json.loads(payload["metadata"])

    assert (
        metadata.get("attempted_retries") == 2
    ), f"Expected attempted_retries=2, got {metadata.get('attempted_retries')}"
    assert (
        metadata.get("max_retries") == 3
    ), f"Expected max_retries=3, got {metadata.get('max_retries')}"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_handles_missing_retry_info_gracefully():
    """
    Test that retry fields are None when not present in metadata (backward compatibility).
    """
    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
            }
        },
        "standard_logging_object": StandardLoggingPayload(
            id="test-no-retry-456",
            call_type="completion",
            stream=False,
            response_cost=0.001,
            status="success",
            total_tokens=100,
            prompt_tokens=50,
            completion_tokens=50,
            startTime=1234567890.0,
            endTime=1234567891.0,
            completionStartTime=None,
            model_map_information=StandardLoggingModelInformation(
                model_map_key="gpt-3.5-turbo", model_map_value=None
            ),
            model="gpt-3.5-turbo",
            model_id="model-123",
            model_group="openai",
            custom_llm_provider="openai",
            api_base="https://api.openai.com",
            metadata=StandardLoggingMetadata(
                user_api_key_hash="test_hash",
                user_api_key_alias=None,
                user_api_key_team_id=None,
                user_api_key_org_id=None,
                user_api_key_user_id=None,
                user_api_key_team_alias=None,
                spend_logs_metadata=None,
                requester_ip_address=None,
                requester_metadata=None,
                user_api_key_end_user_id=None,
            ),
            cache_hit=False,
            cache_key=None,
            saved_cache_cost=0.0,
            request_tags=[],
            end_user=None,
            requester_ip_address=None,
            messages=[],
            response={},
            error_str=None,
            model_parameters={},
            hidden_params=StandardLoggingHiddenParams(
                model_id="model-123",
                cache_key=None,
                api_base="https://api.openai.com",
                response_cost="0.001",
                litellm_overhead_time_ms=None,
                additional_headers=None,
                batch_models=None,
                litellm_model_name=None,
                usage_object=None,
            ),
        ),
    }

    response_obj = {
        "id": "test-response-no-retry",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    metadata = json.loads(payload["metadata"])

    assert (
        metadata.get("attempted_retries") is None
    ), "attempted_retries should be None when not provided"
    assert (
        metadata.get("max_retries") is None
    ), "max_retries should be None when not provided"


def test_get_request_duration_ms_normal():
    """Test that request duration is correctly computed in milliseconds."""
    start = datetime.datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
    end = datetime.datetime(
        2025, 1, 1, 0, 0, 2, 500000, tzinfo=timezone.utc
    )  # 2.5s later
    result = _get_request_duration_ms(start, end)
    assert result == 2500


def test_get_request_duration_ms_sub_millisecond():
    """Test that sub-millisecond durations are truncated to int."""
    start = datetime.datetime(2025, 1, 1, 0, 0, 0, 0, tzinfo=timezone.utc)
    end = datetime.datetime(2025, 1, 1, 0, 0, 0, 500, tzinfo=timezone.utc)  # 0.5ms
    result = _get_request_duration_ms(start, end)
    assert result == 0


def test_get_request_duration_ms_zero():
    """Test that identical start and end times produce 0."""
    t = datetime.datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
    result = _get_request_duration_ms(t, t)
    assert result == 0


def test_get_logging_payload_includes_request_duration_ms():
    """Test that get_logging_payload populates request_duration_ms."""
    start_time = datetime.datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
    end_time = datetime.datetime(2025, 1, 1, 0, 0, 3, tzinfo=timezone.utc)  # 3s later

    kwargs = {
        "model": "gpt-4",
        "litellm_params": {"api_base": "https://api.openai.com"},
        "standard_logging_object": None,
    }
    response_obj = {
        "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
    }

    with (
        patch("litellm.proxy.proxy_server.master_key", None),
        patch("litellm.proxy.proxy_server.general_settings", {}),
    ):
        payload = get_logging_payload(
            kwargs=kwargs,
            response_obj=response_obj,
            start_time=start_time,
            end_time=end_time,
        )

    assert payload["request_duration_ms"] == 3000


class TestIsMasterKey:
    """Tests for _is_master_key handling None inputs without raising TypeError."""

    def test_none_api_key_returns_false(self):
        """Regression: _is_master_key(None, 'sk-master') should return False, not raise TypeError."""
        assert _is_master_key(api_key=None, _master_key="sk-master-key") is False

    def test_none_master_key_returns_false(self):
        assert _is_master_key(api_key="sk-some-key", _master_key=None) is False

    def test_both_none_returns_false(self):
        assert _is_master_key(api_key=None, _master_key=None) is False

    def test_matching_key_returns_true(self):
        assert _is_master_key(api_key="sk-master", _master_key="sk-master") is True

    def test_non_matching_key_returns_false(self):
        assert _is_master_key(api_key="sk-other", _master_key="sk-master") is False

    def test_master_key_hash_is_rejected(self):
        """
        ``_is_master_key`` must not accept ``hash_token(master_key)`` as
        equivalent to the raw master key — only the raw value matches.
        """
        from litellm.proxy.utils import hash_token

        master = "sk-master-key-123"
        hashed = hash_token(master)
        assert _is_master_key(api_key=hashed, _master_key=master) is False


def test_sanitize_request_body_strips_secret_fields():
    """
    secret_fields contains raw HTTP headers (including Authorization Bearer
    tokens). _sanitize_request_body_for_spend_logs_payload must strip it so
    that sensitive credentials are never persisted in the spend-logs DB.
    """
    request_body = {
        "model": "gpt-4",
        "messages": [{"role": "user", "content": "hi"}],
        "secret_fields": {
            "raw_headers": {
                "authorization": "Bearer sk-GjX--WwRQmiX2cvASbKf5Q",
                "content-type": "application/json",
                "host": "litellm.example.com",
            }
        },
    }
    sanitized = _sanitize_request_body_for_spend_logs_payload(request_body)

    assert (
        "secret_fields" not in sanitized
    ), "secret_fields must be stripped from the sanitized request body"
    assert sanitized["model"] == "gpt-4"
    assert sanitized["messages"] == [{"role": "user", "content": "hi"}]


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_proxy_server_request_payload_excludes_secret_fields(mock_should_store):
    """
    End-to-end test: when the proxy_server_request body contains
    secret_fields (as it did before the body-snapshot fix), the spend-log
    serialization must still strip them via the sanitizer fallback.
    """
    mock_should_store.return_value = True

    litellm_params = {
        "proxy_server_request": {
            "body": {
                "model": "gpt-4",
                "messages": [{"role": "user", "content": "hello"}],
                "secret_fields": {
                    "raw_headers": {
                        "authorization": "Bearer sk-super-secret-token",
                        "host": "litellm.example.com",
                    }
                },
            }
        }
    }

    result = _get_proxy_server_request_for_spend_logs_payload(
        metadata={}, litellm_params=litellm_params, kwargs={}
    )
    parsed = json.loads(result)

    assert (
        "secret_fields" not in parsed
    ), "secret_fields must never appear in the spend-log proxy_server_request column"
    assert parsed["model"] == "gpt-4"
    assert parsed["messages"] == [{"role": "user", "content": "hello"}]


# ---------------------------------------------------------------------------
# LIT-2992: error_information sanitization for spend logs
# ---------------------------------------------------------------------------


def test_redact_prompt_leaks_strips_input_value_python_repr():
    # OpenAI-style pydantic validation error: each entry carries its own
    # 'input': [...] field echoing the full conversation.
    error_text = (
        "OpenAIException - {'error': {'message': \"1 validation error:\\n  "
        "{'type': 'string_type', 'loc': ('body', 'input', 'str'), "
        "'msg': 'Input should be a valid string', "
        "'input': [{'role': 'user', 'content': 'super-secret-prompt'}]}"
        '"}}'
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "super-secret-prompt" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted
    # Surrounding context (error class, msg, loc) is preserved.
    assert "string_type" in redacted
    assert "Input should be a valid string" in redacted


def test_redact_prompt_leaks_strips_input_value_json():
    error_text = (
        '{"error":{"message":"validation failed",'
        '"input":[{"role":"user","content":"top-secret-content"}]}}'
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "top-secret-content" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_strips_messages_value():
    error_text = '{"error":{"messages":[{"role":"user","content":"leak"}]}}'
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leak" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_preserves_prose_mentions():
    # The word "input" / "messages" in prose (not as a key) must not be
    # redacted — only quoted-key matches.
    error_text = "Rate limit exceeded. Reduce input size and retry."
    assert _redact_prompt_leaks_in_error_string(error_text) == error_text


def test_redact_prompt_leaks_empty_string():
    assert _redact_prompt_leaks_in_error_string("") == ""


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_when_not_storing_prompts(
    mock_should_store,
):
    mock_should_store.return_value = False

    error_info = {
        "error_code": "429",
        "error_class": "RateLimitError",
        "llm_provider": "openai",
        "traceback": "Traceback (most recent call last):\n  File ...",
        "error_message": (
            'OpenAIException - {"error":{"message":"validation failed",'
            '"input":[{"role":"user","content":"leaked-prompt-content"}]}}'
        ),
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert "leaked-prompt-content" not in sanitized["error_message"]
    assert REDACTED_BY_LITELM_STRING in sanitized["error_message"]
    # Non-leaking fields untouched.
    assert sanitized["error_code"] == "429"
    assert sanitized["error_class"] == "RateLimitError"
    assert sanitized["llm_provider"] == "openai"


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_skips_redaction_when_storing_prompts(
    mock_should_store,
):
    mock_should_store.return_value = True

    error_info = {
        "error_code": "429",
        "error_class": "RateLimitError",
        "llm_provider": "openai",
        "traceback": "",
        "error_message": (
            'OpenAIException - {"error":{"input":[{"role":"user","content":"kept"}]}}'
        ),
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    # User opted in via store_prompts_in_spend_logs — no key-level redaction.
    assert "kept" in sanitized["error_message"]
    assert REDACTED_BY_LITELM_STRING not in sanitized["error_message"]


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_caps_size_regardless_of_prompt_flag(
    mock_should_store,
):
    # The DB-storage cap must apply even when prompt storage is enabled, so a
    # provider error that echoes a multi-MB body can't blow up a single row.
    from litellm.constants import MAX_STRING_LENGTH_PROMPT_IN_DB

    mock_should_store.return_value = True

    huge_error = "x" * (MAX_STRING_LENGTH_PROMPT_IN_DB * 10)
    error_info = {
        "error_code": "500",
        "error_class": "InternalServerError",
        "llm_provider": "openai",
        "traceback": "x" * (MAX_STRING_LENGTH_PROMPT_IN_DB * 10),
        "error_message": huge_error,
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert len(sanitized["error_message"]) < len(huge_error)
    assert LITELLM_TRUNCATED_PAYLOAD_FIELD in sanitized["error_message"]
    assert LITELLM_TRUNCATED_PAYLOAD_FIELD in sanitized["traceback"]


def test_sanitize_error_information_none_passthrough():
    assert _sanitize_error_information_for_spend_logs(None) is None


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_reproduces_lit_2992(mock_should_store):
    # Mirrors the reproduced row body from LIT-2992 — a RateLimitError whose
    # message embeds 178 pydantic validation errors, each carrying a full
    # 'input': [...] echo of the conversation.
    mock_should_store.return_value = False

    huge_conversation_blob = "user-conversation-history-" * 5000
    validation_entries = []
    for _ in range(50):
        validation_entries.append(
            "{'type': 'string_type', 'loc': ('body', 'input', 'str'), "
            "'msg': 'Input should be a valid string', "
            f"'input': [{{'role': 'user', 'content': '{huge_conversation_blob}'}}]}}"
        )
    error_message = (
        "litellm.RateLimitError: RateLimitError: OpenAIException - "
        '{"error":{"message":"' + "\\n  ".join(validation_entries) + '"}}'
    )

    error_info = {
        "error_code": "429",
        "error_class": "RateLimitError",
        "llm_provider": "openai",
        "traceback": "",
        "error_message": error_message,
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert huge_conversation_blob not in sanitized["error_message"]
    # The structural fields that aid debugging remain.
    assert "RateLimitError" in sanitized["error_message"]
    assert "string_type" in sanitized["error_message"]


def test_redact_prompt_leaks_handles_nested_multimodal_content():
    # Multi-modal payload: 'content' is itself a list. The depth-1 regex
    # would stop at the inner '['; the parser-based scanner must walk
    # through balanced nested brackets.
    error_text = (
        '{"error":{"messages":[{"role":"user",'
        '"content":[{"type":"text","text":"top-secret-multimodal"}]}]}}'
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "top-secret-multimodal" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_handles_bracket_in_prompt_text():
    # Prompt text contains a literal '[' — the depth-1 regex would close
    # the outer ']' prematurely. The parser must respect string quoting.
    error_text = (
        '{"error":{"input":[{"role":"user","content":"secret[123 still secret"}]}}'
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "secret[123" not in redacted
    assert "still secret" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_handles_escaped_quote_in_prompt_text():
    # Prompt with an escaped quote inside a JSON string must not break
    # value scanning.
    error_text = '{"error":{"input":[{"role":"user","content":"she said \\"hi[\\""}]}}'
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "she said" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_handles_nested_input_python_repr():
    # Python dict-repr with nested list inside 'input' — single quotes.
    error_text = (
        "validation error: {'input': [{'role': 'user', "
        "'content': [{'type': 'text', 'text': 'leaked-nested-text'}]}]}"
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leaked-nested-text" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_handles_unterminated_value():
    # If a value never closes (malformed error string), redact through to
    # the end rather than leaving the prompt content reachable.
    error_text = '{"input":[{"role":"user","content":"never-closes-leaked'
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "never-closes-leaked" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_traceback_when_not_storing_prompts(
    mock_should_store,
):
    # If a Python exception bubbles up with the request body embedded in
    # its repr (e.g. ValueError(f"bad request: {body}")), the traceback
    # column would carry the prompt unredacted. Verify the redaction
    # covers the traceback field too.
    mock_should_store.return_value = False

    error_info = {
        "error_code": "500",
        "error_class": "ValueError",
        "llm_provider": "",
        "traceback": (
            'Traceback (most recent call last):\n  File "x.py", line 1, in <module>\n'
            '    raise ValueError({"input":[{"role":"user","content":"tb-leaked-prompt"}]})\n'
            "ValueError: invalid request"
        ),
        "error_message": "invalid request",
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert "tb-leaked-prompt" not in sanitized["traceback"]
    assert REDACTED_BY_LITELM_STRING in sanitized["traceback"]
    # Surrounding traceback frames remain so the error stays debuggable.
    assert "Traceback (most recent call last):" in sanitized["traceback"]
    assert "ValueError: invalid request" in sanitized["traceback"]


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_skips_traceback_redaction_when_storing_prompts(
    mock_should_store,
):
    mock_should_store.return_value = True

    error_info = {
        "error_code": "500",
        "error_class": "ValueError",
        "llm_provider": "",
        "traceback": (
            'raise ValueError({"input":[{"role":"user","content":"tb-kept"}]})'
        ),
        "error_message": "invalid request",
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert "tb-kept" in sanitized["traceback"]
    assert REDACTED_BY_LITELM_STRING not in sanitized["traceback"]


def test_redact_prompt_leaks_strips_prompt_key_completions_payload():
    # /v1/completions echoes the user input under the top-level 'prompt' key
    # rather than 'messages'. Without 'prompt' coverage the body would survive
    # the redactor when store_prompts_in_spend_logs is False.
    error_text = (
        '{"error":{"message":"validation failed",'
        '"prompt":"super-secret-completion-text"}}'
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "super-secret-completion-text" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_strips_prompt_key_python_repr():
    error_text = (
        "{'model': 'gpt-3.5-turbo-instruct', "
        "'prompt': 'leaked-completion-prompt-body'}"
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leaked-completion-prompt-body" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_preserves_prompt_substring_keys():
    # 'prompt_tokens' / 'prompt_token_count' / etc. are not the leak key —
    # the matcher requires a closing quote before ':' so substrings shouldn't
    # trigger redaction.
    error_text = '{"usage":{"prompt_tokens":42,"completion_tokens":7}}'
    assert _redact_prompt_leaks_in_error_string(error_text) == error_text


def test_redact_prompt_leaks_strips_pydantic_input_value_list():
    # Pydantic v2 validation error format — the offending value is rendered
    # as a Python repr after `input_value=`. The full repr can carry the
    # entire request body and must be redacted under the same gate as the
    # quoted-key form.
    error_text = (
        "1 validation error for ChatCompletionRequest\n"
        "messages\n"
        "  Input should be a valid list "
        "[type=list_type, input_value=['secret-pydantic-prompt'], input_type=str]"
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "secret-pydantic-prompt" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted
    # Surrounding pydantic context is preserved so the error stays debuggable.
    assert "list_type" in redacted
    assert "input_type=str" in redacted


def test_redact_prompt_leaks_strips_pydantic_input_value_dict():
    error_text = (
        "[type=dict_type, "
        "input_value={'role': 'user', 'content': 'leaked-dict-content'}, "
        "input_type=dict]"
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leaked-dict-content" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted
    assert "input_type=dict" in redacted


def test_redact_prompt_leaks_strips_pydantic_input_value_quoted_string():
    error_text = "[type=string_type, input_value='leaked-string-value', input_type=str]"
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leaked-string-value" not in redacted
    assert REDACTED_BY_LITELM_STRING in redacted


def test_redact_prompt_leaks_pydantic_input_value_scalar_left_intact():
    # Bare numeric / bool / None scalars in input_value are not prompt
    # carriers; leaving them untouched keeps the validation error readable.
    error_text = "[type=int_type, input_value=42, input_type=int]"
    assert _redact_prompt_leaks_in_error_string(error_text) == error_text


def test_redact_prompt_leaks_input_value_substring_does_not_match():
    # Word-boundary anchoring on `input_value` so similarly named keys (e.g.
    # `my_input_value=` from another stack frame) are not mis-redacted.
    error_text = "frame: my_input_value=42 elsewhere"
    assert _redact_prompt_leaks_in_error_string(error_text) == error_text


def test_redact_prompt_leaks_combined_quoted_key_and_pydantic_assignment():
    # A real OpenAI/pydantic error often carries BOTH forms in the same
    # string — quoted JSON 'input' echoed once, then pydantic 'input_value='
    # echoed per validation entry. Both must be redacted in one pass.
    error_text = (
        '{"error":{"input":[{"role":"user","content":"leak-via-json"}]}} '
        "[type=list_type, input_value=['leak-via-pydantic'], input_type=str]"
    )
    redacted = _redact_prompt_leaks_in_error_string(error_text)
    assert "leak-via-json" not in redacted
    assert "leak-via-pydantic" not in redacted
    assert redacted.count(REDACTED_BY_LITELM_STRING) >= 2


@patch(
    "litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs"
)
def test_sanitize_error_information_redacts_pydantic_assignment_form(
    mock_should_store,
):
    # End-to-end: a pydantic-style error that lands in error_message must be
    # redacted under the spend-log path, not just the regex-level helper.
    mock_should_store.return_value = False

    error_info = {
        "error_code": "422",
        "error_class": "ValidationError",
        "llm_provider": "openai",
        "traceback": "",
        "error_message": (
            "1 validation error for ChatCompletionRequest\n"
            "messages\n"
            "  Field required "
            "[type=missing, input_value={'prompt': 'leaked-via-pydantic-msg'}, "
            "input_type=dict]"
        ),
    }

    sanitized = _sanitize_error_information_for_spend_logs(error_info)

    assert sanitized is not None
    assert "leaked-via-pydantic-msg" not in sanitized["error_message"]
    assert REDACTED_BY_LITELM_STRING in sanitized["error_message"]


# ── _redact_logged_api_key unit tests ──────────────────────────────────────


def test_redact_logged_api_key_none_returns_none():
    assert _redact_logged_api_key(None) is None


def test_redact_logged_api_key_empty_string_returns_none():
    assert _redact_logged_api_key("") is None


def test_redact_logged_api_key_sk_key_is_hashed():
    raw = "sk-1234secret"
    result = _redact_logged_api_key(raw)
    assert result == hash_token(raw)
    assert result is not None
    assert not result.startswith("sk-")
    assert len(result) == 64


def test_redact_logged_api_key_bearer_sk_equals_sk_hash():
    raw = "sk-1234secret"
    result_plain = _redact_logged_api_key(raw)
    result_bearer = _redact_logged_api_key(f"Bearer {raw}")
    assert result_bearer == result_plain


def test_redact_logged_api_key_bearer_case_insensitive():
    raw = "sk-1234secret"
    result_lower = _redact_logged_api_key(f"bearer {raw}")
    result_upper = _redact_logged_api_key(f"BEARER {raw}")
    expected = hash_token(raw)
    assert result_lower == expected
    assert result_upper == expected


def test_redact_logged_api_key_non_sk_raw_key_is_hashed():
    raw = "anthropic-raw-key-xyz"
    result = _redact_logged_api_key(raw)
    assert result is not None
    assert result != raw
    assert len(result) == 64
    assert result == hash_token(raw)


def test_redact_logged_api_key_already_valid_sha256_passes_through_with_flag():
    already_hashed = hash_token("sk-some-key")
    assert len(already_hashed) == 64
    result = _redact_logged_api_key(already_hashed, already_redacted=True)
    assert result == already_hashed
    assert hash_token(already_hashed) != result  # no double-hash


def test_redact_logged_api_key_sha256_without_flag_is_hashed():
    already_hashed = hash_token("sk-some-key")
    assert len(already_hashed) == 64
    result = _redact_logged_api_key(already_hashed)
    assert result is not None
    assert result != already_hashed
    assert len(result) == 64
    assert result == hash_token(already_hashed)


def test_redact_logged_api_key_long_opaque_token_is_hashed():
    raw = "x1" * 450
    assert len(raw) == 900
    result = _redact_logged_api_key(raw)
    assert result is not None
    assert result != raw
    assert raw not in result
    assert len(result) == 64
    assert result == hash_token(raw)


def test_redact_logged_api_key_hashed_jwt_passes_through():
    jwt_hash = "hashed-jwt-" + "a" * 64
    result = _redact_logged_api_key(jwt_hash, already_redacted=True)
    assert result == jwt_hash


def test_redact_logged_api_key_hashed_jwt_shape_without_provenance_is_hashed():
    lookalike = "hashed-jwt-" + "a" * 64
    result = _redact_logged_api_key(lookalike)
    assert result == hash_token(lookalike)
    assert result != lookalike


def test_redact_logged_api_key_hashed_jwt_trailing_newline_is_hashed():
    trailing = "hashed-jwt-" + "a" * 64 + "\n"
    result = _redact_logged_api_key(trailing, already_redacted=True)
    assert result == hash_token(trailing)
    assert result != trailing


def test_redact_logged_api_key_hashed_jwt_short_suffix_is_hashed():
    short_jwt = "hashed-jwt-tooshort"
    result = _redact_logged_api_key(short_jwt)
    assert result is not None
    assert result != short_jwt
    assert len(result) == 64
    assert result == hash_token(short_jwt)


def test_redact_logged_api_key_master_key_alias_passes_through():
    from litellm.constants import LITELLM_PROXY_MASTER_KEY_ALIAS

    result = _redact_logged_api_key(LITELLM_PROXY_MASTER_KEY_ALIAS, already_redacted=True)
    assert result == LITELLM_PROXY_MASTER_KEY_ALIAS


def test_redact_logged_api_key_master_key_alias_without_provenance_is_hashed():
    from litellm.constants import LITELLM_PROXY_MASTER_KEY_ALIAS

    result = _redact_logged_api_key(LITELLM_PROXY_MASTER_KEY_ALIAS)
    assert result == hash_token(LITELLM_PROXY_MASTER_KEY_ALIAS)
    assert result != LITELLM_PROXY_MASTER_KEY_ALIAS


def test_get_spend_logs_metadata_keeps_master_key_alias_readable():
    from litellm.constants import LITELLM_PROXY_MASTER_KEY_ALIAS

    meta = _get_spend_logs_metadata(
        {
            "user_api_key": LITELLM_PROXY_MASTER_KEY_ALIAS,
            "user_api_key_hash": LITELLM_PROXY_MASTER_KEY_ALIAS,
        }
    )
    assert meta["user_api_key"] == LITELLM_PROXY_MASTER_KEY_ALIAS


def test_redact_logged_api_key_bearer_only_returns_none():
    # "bearer " with nothing after stripping is equivalent to no key
    assert _redact_logged_api_key("bearer ") is None
    assert _redact_logged_api_key("Bearer ") is None
    assert _redact_logged_api_key("BEARER ") is None


# ── _get_spend_logs_metadata key-hash invariant tests ─────────────────────


def test_get_spend_logs_metadata_sk_key_hashed():
    raw = "sk-1234secret"
    meta = _get_spend_logs_metadata({"user_api_key": raw})
    assert meta["user_api_key"] == hash_token(raw)
    assert meta["user_api_key"] is not None
    result = meta["user_api_key"]
    assert result is not None
    assert not result.startswith("sk-")
    assert len(result) == 64


def test_get_spend_logs_metadata_bearer_sk_key_hashed_same_as_plain():
    raw = "sk-1234secret"
    meta_plain = _get_spend_logs_metadata({"user_api_key": raw})
    meta_bearer = _get_spend_logs_metadata({"user_api_key": f"Bearer {raw}"})
    assert meta_bearer["user_api_key"] == meta_plain["user_api_key"]


def test_get_spend_logs_metadata_non_sk_raw_key_hashed():
    raw = "anthropic-raw-key-xyz"
    meta = _get_spend_logs_metadata({"user_api_key": raw})
    result = meta["user_api_key"]
    assert result is not None
    assert result != raw
    assert len(result) == 64


def test_get_spend_logs_metadata_already_hashed_unchanged_with_provenance():
    already_hashed = hash_token("sk-some-key")
    meta = _get_spend_logs_metadata(
        {"user_api_key": already_hashed, "user_api_key_hash": already_hashed}
    )
    assert meta["user_api_key"] == already_hashed
    assert hash_token(already_hashed) != meta["user_api_key"]  # no double-hash


def test_get_spend_logs_metadata_already_hashed_no_provenance_is_rehashed():
    already_hashed = hash_token("sk-some-key")
    meta = _get_spend_logs_metadata({"user_api_key": already_hashed})
    assert meta["user_api_key"] != already_hashed
    assert meta["user_api_key"] == hash_token(already_hashed)


def test_get_spend_logs_metadata_provenance_bypass_requires_hash_match():
    already_hashed = hash_token("sk-some-key")
    different_hash = hash_token("sk-other-key")
    meta = _get_spend_logs_metadata(
        {"user_api_key": already_hashed, "user_api_key_hash": different_hash}
    )
    assert meta["user_api_key"] == hash_token(already_hashed)


def test_get_spend_logs_metadata_hashed_jwt_unchanged():
    jwt_hash = "hashed-jwt-" + "b" * 64
    meta = _get_spend_logs_metadata({"user_api_key": jwt_hash, "user_api_key_hash": jwt_hash})
    assert meta["user_api_key"] == jwt_hash


def test_get_spend_logs_metadata_hashed_jwt_shape_without_provenance_is_hashed():
    lookalike = "hashed-jwt-" + "b" * 64
    meta = _get_spend_logs_metadata({"user_api_key": lookalike})
    assert meta["user_api_key"] == hash_token(lookalike)
    assert meta["user_api_key"] != lookalike


def test_get_spend_logs_metadata_none_key_is_none():
    meta = _get_spend_logs_metadata({"user_api_key": None})
    assert meta["user_api_key"] is None


# ── get_logging_payload key-hash invariant tests ───────────────────────────


def test_get_logging_payload_uses_recovered_combined_usage_on_failure():
    """A request that fails mid-stream has no usable response_obj usage, but the
    streaming handler recovers the usage from the chunks already delivered and
    the failure hook surfaces it as ``combined_usage_object``. The spend-log
    payload must record those token counts instead of zero.
    """
    from litellm.types.utils import Usage

    kwargs = {
        "model": "anthropic/claude-haiku-4-5",
        "call_type": "acompletion",
        "litellm_params": {"metadata": {"user_api_key": "sk-test"}},
        "combined_usage_object": Usage(
            prompt_tokens=30, completion_tokens=1, total_tokens=31
        ),
    }
    response_obj = Exception("MidStreamFallbackError: read timeout")
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )

    assert payload["prompt_tokens"] == 30
    assert payload["completion_tokens"] == 1
    assert payload["total_tokens"] == 31


def test_get_logging_payload_failure_without_recovered_usage_is_zero():
    """A failure with no recovered usage keeps zero token counts, so the
    combined-usage override never invents tokens for ordinary failures.
    """
    kwargs = {
        "model": "anthropic/claude-haiku-4-5",
        "call_type": "acompletion",
        "litellm_params": {"metadata": {"user_api_key": "sk-test"}},
    }
    response_obj = Exception("BadRequestError")
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )

    assert payload["total_tokens"] == 0


def test_get_logging_payload_sets_litellm_call_id_for_correlation():
    """LIT-3868: a successful spend log must carry the x-litellm-call-id (the
    trace id) in its metadata, distinct from request_id, which stays the
    provider response id. Without this there is no way to correlate a DB row
    with its trace for a successful call.
    """
    provider_response_id = "chatcmpl-e6e6f3e9-c392-404e-9a71-5361c79d8470"
    trace_call_id = "c6a77556-19ce-4406-b287-53f5fb4b2b55"

    kwargs = {
        "model": "openai/gpt-4o-mini",
        "call_type": "acompletion",
        "litellm_call_id": trace_call_id,
        "litellm_params": {"metadata": {"user_api_key": "sk-test"}},
    }
    response_obj = {
        "id": provider_response_id,
        "usage": {"prompt_tokens": 5, "completion_tokens": 7, "total_tokens": 12},
    }
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )
    metadata = json.loads(payload["metadata"])

    assert payload["request_id"] == provider_response_id
    assert metadata["litellm_call_id"] == trace_call_id
    assert metadata["litellm_call_id"] != payload["request_id"]


def test_get_logging_payload_litellm_call_id_falls_back_to_litellm_params():
    """litellm_call_id may only be present in litellm_params; it must still land
    in the spend log metadata so correlation works on that path too.
    """
    trace_call_id = "fallback-7a1c-42d9-9f0e-2b6c5d4e3f21"
    kwargs = {
        "model": "openai/gpt-4o-mini",
        "call_type": "acompletion",
        "litellm_params": {
            "litellm_call_id": trace_call_id,
            "metadata": {"user_api_key": "sk-test"},
        },
    }
    response_obj = {
        "id": "chatcmpl-abc123",
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
    }
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )

    assert json.loads(payload["metadata"])["litellm_call_id"] == trace_call_id


def test_get_logging_payload_litellm_call_id_when_response_has_no_id():
    """When the provider returns no id, request_id falls back to the call id, so
    request_id and the metadata call id hold the same value and correlation
    still resolves.
    """
    trace_call_id = "noid-5b2e-4c7a-9d10-3f8a1c2b4e6d"
    kwargs = {
        "model": "openai/gpt-4o-mini",
        "call_type": "acompletion",
        "litellm_call_id": trace_call_id,
        "litellm_params": {"metadata": {"user_api_key": "sk-test"}},
    }
    response_obj = {
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
    }
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )

    assert json.loads(payload["metadata"])["litellm_call_id"] == trace_call_id
    assert payload["request_id"] == trace_call_id


def test_get_logging_payload_cache_hit_keeps_raw_litellm_call_id():
    """On a cache hit request_id is suffixed to stay unique, but the metadata
    litellm_call_id stays the raw trace id so the row still points at its trace.
    """
    trace_call_id = "cache-9a1c-42d9-9f0e-2b6c5d4e3f21"
    kwargs = {
        "model": "openai/gpt-4o-mini",
        "call_type": "acompletion",
        "litellm_call_id": trace_call_id,
        "cache_hit": True,
        "litellm_params": {"metadata": {"user_api_key": "sk-test"}},
    }
    response_obj = {
        "id": "chatcmpl-cache-src",
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
    }
    now = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs, response_obj=response_obj, start_time=now, end_time=now
    )

    assert json.loads(payload["metadata"])["litellm_call_id"] == trace_call_id
    assert "_cache_hit" in payload["request_id"]
    assert json.loads(payload["metadata"])["litellm_call_id"] != payload["request_id"]


class TestSpendLogKeyRedaction:
    """Regression: plaintext API keys with Bearer prefix were stored in
    SpendLogs for failed requests (LIT-4121)"""

    def test_bearer_prefixed_sk_key_is_hashed(self):
        raw = "Bearer sk-WLi4iRn4JmbVlTaYw12IOA"
        result = _redact_logged_api_key(raw)
        assert result is not None
        assert not result.startswith("Bearer")
        assert not result.startswith("sk-")
        assert len(result) == 64

    def test_bare_sk_key_is_hashed(self):
        raw = "sk-WLi4iRn4JmbVlTaYw12IOA"
        result = _redact_logged_api_key(raw)
        assert result is not None
        assert not result.startswith("sk-")
        assert len(result) == 64

    def test_bearer_lowercase_is_handled(self):
        raw = "bearer sk-WLi4iRn4JmbVlTaYw12IOA"
        result = _redact_logged_api_key(raw)
        assert result is not None
        assert not result.startswith("bearer")
        assert not result.startswith("sk-")
        assert len(result) == 64

    def test_already_hashed_key_unchanged(self):
        hashed = "bcfe8173f5447f10be0e7fb37aaa8b97829d5c9e0498232152f9d123456789ab"
        assert _redact_logged_api_key(hashed, already_redacted=True) == hashed

    def test_bearer_prefixed_non_sk_key_is_hashed(self):
        raw = "Bearer some-other-token-format"
        result = _redact_logged_api_key(raw)
        assert result == hash_token("some-other-token-format")
        assert result is not None
        assert not result.startswith("Bearer")

    def test_bearer_and_bare_produce_same_hash(self):
        bare = "sk-WLi4iRn4JmbVlTaYw12IOA"
        bearer = "Bearer sk-WLi4iRn4JmbVlTaYw12IOA"
        assert _redact_logged_api_key(bare) == _redact_logged_api_key(bearer)


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_non_sk_raw_key_both_fields_hashed():
    raw = "anthropic-raw-key-xyz"
    kwargs = {
        "model": "openai/gpt-4.1",
        "messages": [{"role": "user", "content": "Hello"}],
        "call_type": "acompletion",
        "litellm_params": {
            "metadata": {
                "user_api_key": raw,
                "user_api_key_user_id": "test_user",
                "user_api_key_team_id": "test_team",
            }
        },
    }
    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=Exception("error"),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )

    assert payload["api_key"] != raw
    assert len(payload["api_key"]) == 64

    parsed_meta = json.loads(payload["metadata"])
    assert parsed_meta["user_api_key"] != raw
    assert parsed_meta["user_api_key"] is not None
    assert len(parsed_meta["user_api_key"]) == 64


def test_get_logging_payload_keeps_master_key_alias_readable():
    from litellm.constants import LITELLM_PROXY_MASTER_KEY_ALIAS

    kwargs = {
        "model": "openai/gpt-4.1",
        "messages": [{"role": "user", "content": "Hello"}],
        "call_type": "acompletion",
        "litellm_params": {
            "metadata": {
                "user_api_key": LITELLM_PROXY_MASTER_KEY_ALIAS,
                "user_api_key_hash": LITELLM_PROXY_MASTER_KEY_ALIAS,
                "user_api_key_user_id": "test_user",
            }
        },
    }
    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=Exception("error"),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )

    assert payload["api_key"] == LITELLM_PROXY_MASTER_KEY_ALIAS
    parsed_meta = json.loads(payload["metadata"])
    assert parsed_meta["user_api_key"] == LITELLM_PROXY_MASTER_KEY_ALIAS


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_hashes_bearer_prefixed_api_key():
    """Regression for LIT-4121: failed-request spend logs stored plaintext
    'Bearer sk-...' in both the api_key column and metadata.user_api_key"""
    raw_key = "Bearer sk-WLi4iRn4JmbVlTaYw12IOA"

    kwargs = {
        "model": "openai/gpt-4.1",
        "call_type": "acompletion",
        "litellm_params": {
            "metadata": {
                "user_api_key": raw_key,
                "user_api_key_user_id": "test_user",
                "user_api_key_team_id": "test_team",
                "status": "failure",
            }
        },
    }

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=Exception("model error"),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )

    assert not payload["api_key"].startswith("Bearer"), (
        f"api_key column contains plaintext Bearer key: {payload['api_key']}"
    )
    assert not payload["api_key"].startswith("sk-"), (
        f"api_key column contains unhashed key: {payload['api_key']}"
    )

    metadata_dict = json.loads(payload["metadata"])
    assert not metadata_dict["user_api_key"].startswith("Bearer"), (
        f"metadata user_api_key contains plaintext Bearer key: {metadata_dict['user_api_key']}"
    )
    assert not metadata_dict["user_api_key"].startswith("sk-"), (
        f"metadata user_api_key contains unhashed key: {metadata_dict['user_api_key']}"
    )


@patch("litellm.proxy.spend_tracking.spend_tracking_utils._should_store_prompts_and_responses_in_spend_logs")
def test_sanitize_guardrail_information_preserves_headroom_compression_token_stats(
    mock_should_store,
):
    """
    Headroom's compression stats live in guardrail_response, which is redacted
    by default. Purely numeric token stats carry no prompt content and must
    survive so daily spend aggregation can count compression_saved_tokens;
    everything else in guardrail_response stays redacted.
    """
    mock_should_store.return_value = False
    guardrail_info = [
        {
            "guardrail_name": "headroom-compressor",
            "guardrail_provider": "headroom",
            "guardrail_status": "success",
            "guardrail_response": {
                "tokens_before": 1000,
                "tokens_after": 400,
                "tokens_saved": 600,
                "compression_ratio": 0.4,
                "transforms_applied": ["dedupe_messages"],
            },
        },
        {
            "guardrail_name": "echo-guard",
            "guardrail_status": "success",
            "guardrail_response": {
                "evaluated_input": "secret prompt",
                "tokens_saved": "prompt text hiding in a stat key",
            },
        },
    ]

    result = _sanitize_guardrail_information_for_spend_logs(guardrail_info)

    assert result is not None
    assert result[0]["guardrail_response"] == {
        "tokens_before": 1000,
        "tokens_after": 400,
        "tokens_saved": 600,
        "compression_ratio": 0.4,
    }
    assert result[1]["guardrail_response"] == REDACTED_BY_LITELM_STRING


@pytest.mark.asyncio
async def test_compression_savings_survive_to_spend_log_payload_metadata(monkeypatch):
    """
    End-to-end seam test for the native compression write path on
    /v1/messages: the pre-call deployment hook records savings into the call
    kwargs' litellm_metadata, the logging object captures it via
    update_from_kwargs (exactly as llm_http_handler does for
    anthropic_messages), and get_logging_payload lands it in
    SpendLogsPayload.metadata JSON under ``compression_savings``.
    """
    from litellm.integrations.compression_interception.handler import (
        CompressionInterceptionLogger,
    )
    from litellm.litellm_core_utils.litellm_logging import Logging
    from litellm.types.utils import CallTypes

    monkeypatch.setattr(
        "litellm.integrations.compression_interception.handler.compress",
        lambda **kwargs: {
            "messages": [{"role": "user", "content": "compressed"}],
            "original_tokens": 12000,
            "compressed_tokens": 5000,
            "cache": {"auth.py": "content"},
            "tools": [],
        },
    )
    logger = CompressionInterceptionLogger()

    call_kwargs = {
        "model": "claude-sonnet-5",
        "messages": [{"role": "user", "content": "big context"}],
        "max_tokens": 512,
        "litellm_call_id": "test-compression-call-id",
        "litellm_metadata": {
            "user_api_key": "88dc28d0f030c55ed4ab77ed8faf098196cb1c05df778539800c9f1243fe6b4b",
            "user_api_key_user_id": "u1",
            "user_api_key_team_id": "t1",
        },
    }
    hooked = await logger.async_pre_call_deployment_hook(kwargs=call_kwargs, call_type=CallTypes.anthropic_messages)
    assert hooked is not None

    start_time = datetime.datetime.now(timezone.utc)
    logging_obj = Logging(
        model="claude-sonnet-5",
        messages=hooked["messages"],
        stream=False,
        call_type="anthropic_messages",
        start_time=start_time,
        litellm_call_id="test-compression-call-id",
        function_id="test",
    )
    logging_obj.update_from_kwargs(
        kwargs=hooked,
        model="claude-sonnet-5",
        optional_params={"max_tokens": 512},
        litellm_params={
            "preset_cache_key": None,
            "stream_response": {},
            "model_info": hooked.get("model_info"),
        },
        custom_llm_provider="anthropic",
    )

    end_time = datetime.datetime.now(timezone.utc)
    logging_obj.model_call_details["completion_start_time"] = end_time
    payload = get_logging_payload(
        kwargs=logging_obj.model_call_details,
        response_obj={
            "id": "msg_test",
            "usage": {"prompt_tokens": 5000, "completion_tokens": 10, "total_tokens": 5010},
        },
        start_time=start_time,
        end_time=end_time,
    )

    payload_metadata = json.loads(payload["metadata"])
    assert payload_metadata["compression_savings"] == {
        "tokens_before": 12000,
        "tokens_after": 5000,
        "tokens_saved": 7000,
        "source": "compression_interception",
    }


def test_no_routing_decision_key_defaults_to_none_in_spend_log_metadata():
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-no-routing-decision", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["routing_decision"] is None


@pytest.mark.parametrize("bucket", ["metadata", "litellm_metadata"])
def test_internal_call_origin_survives_into_spend_log_metadata(bucket):
    """The origin is only useful if it reaches the row the Logs UI reads.

    _get_spend_logs_metadata projects onto SpendLogsMetadata.__annotations__, so an
    undeclared key is dropped silently. Both buckets are covered because the resolver
    returns litellm_metadata when present and metadata otherwise, and the classifier
    sub-call populates whichever the parent route used.
    """
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {
                bucket: {
                    "user_api_key": "test-key",
                    "internal_call_origin": "autorouter_classifier",
                }
            },
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-classifier", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["internal_call_origin"] == "autorouter_classifier"


def test_user_traffic_carries_no_internal_call_origin():
    """The negative class the badge depends on: an ordinary request must be
    distinguishable from a classifier call, not merely unlabelled by accident."""
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-user-traffic", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["internal_call_origin"] is None


def _spend_log_for_call_type(
    call_type: str, internal_call_origin: str | None = None, background: bool | None = None
) -> dict:
    from litellm.types.llms.openai import ResponsesAPIResponse

    return cast(
        dict,
        get_logging_payload(
            kwargs={
                "model": "gpt-4o",
                "call_type": call_type,
                "response_cost": 0.0,
                "litellm_params": {
                    "metadata": {
                        "user_api_key": "test-key",
                        "internal_call_origin": internal_call_origin,
                    }
                },
            },
            response_obj=ResponsesAPIResponse(
                id="resp_lit5602",
                created_at=1234567890,
                model="gpt-4o",
                output=[],
                usage={"input_tokens": 4000, "output_tokens": 2000, "total_tokens": 6000},
                background=background,
            ),
            start_time=datetime.datetime.now(timezone.utc),
            end_time=datetime.datetime.now(timezone.utc),
        ),
    )


def test_spend_log_for_response_retrieval_does_not_replay_the_created_responses_tokens():
    """A retrieved response carries the usage of the call that created it, so counting it again
    bills the same tokens twice. Regression test for LIT-5602."""
    payload = _spend_log_for_call_type("aget_responses")

    assert payload["prompt_tokens"] == 0
    assert payload["completion_tokens"] == 0
    assert payload["total_tokens"] == 0
    assert payload["spend"] == 0.0


def test_spend_log_for_background_response_cost_poll_counts_tokens():
    """The poller's read is where a background job's usage first shows up, so dropping it there
    leaves the job unbilled forever."""
    payload = _spend_log_for_call_type("aget_responses", internal_call_origin="background_response_cost_poll")

    assert payload["total_tokens"] == 6000


def test_spend_log_for_background_response_retrieval_counts_tokens():
    """A background create answers queued carrying no usage, so its retrieval is the first and only
    place the job's tokens are ever visible. Zeroing that read bills the whole job nothing on any
    proxy that is not running the enterprise cost poller."""
    payload = _spend_log_for_call_type("aget_responses", background=True)

    assert payload["total_tokens"] == 6000


def test_spend_log_for_foreground_response_retrieval_still_counts_nothing():
    """Guards the test above against a blanket exemption: an explicit background=false read was
    already billed by its create and must stay at zero."""
    payload = _spend_log_for_call_type("aget_responses", background=False)

    assert payload["total_tokens"] == 0


def test_spend_log_for_response_creation_still_counts_tokens():
    """Guards the test above: the same response object must still be counted on the create path."""
    payload = _spend_log_for_call_type("aresponses")

    assert payload["total_tokens"] == 6000


REDACTED_RESPONSE_PLACEHOLDER: Final = {"text": "redacted-by-litellm"}
CONSTANT_ID_FROM_HASHED_PLACEHOLDER: Final = "00fcbef15a3b0097e14b0ca016ed30a0"


@pytest.mark.parametrize("call_type", ["aretrieve_batch", "acreate_file"])
def test_get_spend_logs_id_stays_unique_when_the_response_is_a_redaction_placeholder(call_type):
    """request_id is the LiteLLM_SpendLogs primary key and the flush inserts with
    skip_duplicates, so two calls must never derive the same id from identical response
    content. Message redaction replaces every body it cannot redact with one fixed
    placeholder, which is what a batch and a file body both become, so hashing the
    response collapsed all of them onto a single id and silently dropped every row
    after the first."""
    suffix = "_batch_cost" if call_type == "aretrieve_batch" else ""
    first = get_spend_logs_id(call_type, dict(REDACTED_RESPONSE_PLACEHOLDER), {"litellm_call_id": "call-id-1"})
    second = get_spend_logs_id(call_type, dict(REDACTED_RESPONSE_PLACEHOLDER), {"litellm_call_id": "call-id-2"})

    assert first == f"call-id-1{suffix}"
    assert second == f"call-id-2{suffix}"
    assert first != second
    assert first != CONSTANT_ID_FROM_HASHED_PLACEHOLDER
    assert second != CONSTANT_ID_FROM_HASHED_PLACEHOLDER


@pytest.mark.parametrize("call_type", ["aretrieve_batch", "acreate_file"])
def test_get_spend_logs_id_prefers_the_response_id_for_batch_and_file_calls(call_type):
    """A batch or file response that survives redaction carries its own id, so the row
    keys off that rather than the per-call id."""
    expected = "batch_abc123_batch_cost" if call_type == "aretrieve_batch" else "batch_abc123"
    assert get_spend_logs_id(call_type, {"id": "batch_abc123"}, {"litellm_call_id": "call-id-1"}) == expected


def test_get_logging_payload_gives_redacted_batch_and_file_rows_distinct_request_ids():
    """End to end at the payload level: a batch retrieve and a file create whose bodies
    were both flattened to the same redaction placeholder must still produce two
    insertable rows, each carrying its own spend."""
    payloads = [
        get_logging_payload(
            kwargs={
                "call_type": call_type,
                "model": model,
                "litellm_call_id": call_id,
                "litellm_params": {"metadata": {"user_api_key": "test-key"}},
            },
            response_obj=dict(REDACTED_RESPONSE_PLACEHOLDER),
            start_time=datetime.datetime.now(timezone.utc),
            end_time=datetime.datetime.now(timezone.utc),
        )
        for call_type, model, call_id in (
            ("aretrieve_batch", "global.anthropic.claude-haiku-4-5-20251001-v1:0", "call-id-batch"),
            ("acreate_file", "vertex_ai/gemini-2.5-flash", "call-id-file"),
        )
    ]
    request_ids = [payload["request_id"] for payload in payloads]

    assert request_ids == ["call-id-batch_batch_cost", "call-id-file"]
    assert len(set(request_ids)) == len(request_ids)
    assert CONSTANT_ID_FROM_HASHED_PLACEHOLDER not in request_ids


@pytest.mark.parametrize("call_type", ["aretrieve_batch", "acreate_file"])
def test_get_spend_logs_id_keys_off_batch_identity_when_the_body_was_redacted(call_type):
    """Retrieving one batch twice must produce one row, not two. Redaction strips the id
    off the response body, so the identity has to come from the standard logging payload,
    which is built from the unredacted response and keeps it. Falling through to the
    per-call id here would write a second row carrying the same batch's full cost and
    overstate spend by a multiple of how often the caller polled."""
    standard_logging_object = {"id": "batch_abc123"}
    first = get_spend_logs_id(
        call_type,
        dict(REDACTED_RESPONSE_PLACEHOLDER),
        {"litellm_call_id": "call-id-1", "standard_logging_object": standard_logging_object},
    )
    second = get_spend_logs_id(
        call_type,
        dict(REDACTED_RESPONSE_PLACEHOLDER),
        {"litellm_call_id": "call-id-2", "standard_logging_object": standard_logging_object},
    )

    expected = "batch_abc123_batch_cost" if call_type == "aretrieve_batch" else "batch_abc123"
    assert first == second == expected
    assert first != CONSTANT_ID_FROM_HASHED_PLACEHOLDER


def test_get_spend_logs_id_separates_distinct_batches_whose_bodies_were_both_redacted():
    """The flip side of idempotency: two different batches must not share a row just
    because redaction flattened both bodies to the same placeholder."""
    ids = [
        get_spend_logs_id(
            "aretrieve_batch",
            dict(REDACTED_RESPONSE_PLACEHOLDER),
            {"litellm_call_id": f"call-id-{index}", "standard_logging_object": {"id": batch_id}},
        )
        for index, batch_id in enumerate(("batch_first", "batch_second"))
    ]

    assert ids == ["batch_first_batch_cost", "batch_second_batch_cost"]


def test_get_spend_logs_id_prefers_the_response_id_over_the_standard_logging_id():
    """An unredacted response keeps deciding its own row key, so cache-hit ids and every
    other call type behave exactly as they did before."""
    assert (
        get_spend_logs_id(
            "acompletion",
            {"id": "chatcmpl-from-response"},
            {"litellm_call_id": "call-id-1", "standard_logging_object": {"id": "id-from-standard-payload"}},
        )
        == "chatcmpl-from-response"
    )


def test_batch_cost_row_does_not_collide_with_the_batch_creation_row():
    """Creating a batch writes a row keyed by the batch's own id, so keying the cost row
    the same way makes the insert a duplicate of it. request_id is the primary key and the
    flush skips duplicates, so the cost row is dropped with no error and the batch is
    billed nothing. Observed against a live proxy: the poller computed and flushed the
    cost, and the only row carrying that id was the acreate_batch row written when the
    batch was submitted."""
    batch_id = "bGl0ZWxsbV9wcm94eTttb2RlbF9pZDphYmM7bGxtX2JhdGNoX2lkOnh5eg"

    creation_row_id = get_spend_logs_id("acreate_batch", {"id": batch_id}, {"litellm_call_id": "call-create"})
    cost_row_id = get_spend_logs_id(
        "aretrieve_batch",
        dict(REDACTED_RESPONSE_PLACEHOLDER),
        {"litellm_call_id": "call-poller", "standard_logging_object": {"id": batch_id}},
    )

    assert creation_row_id == batch_id
    assert cost_row_id != creation_row_id
    assert cost_row_id == f"{batch_id}_batch_cost"


def test_batch_cost_row_id_is_stable_across_repeated_accounting():
    """The cost row stays keyed to the batch, so accounting the same batch twice collapses
    to one row instead of billing it twice."""
    standard_logging_object = {"id": "batch_same"}
    ids = [
        get_spend_logs_id(
            "aretrieve_batch",
            dict(REDACTED_RESPONSE_PLACEHOLDER),
            {"litellm_call_id": f"call-{index}", "standard_logging_object": standard_logging_object},
        )
        for index in range(2)
    ]

    assert ids[0] == ids[1] == "batch_same_batch_cost"


def _make_failed_request_standard_logging_payload() -> StandardLoggingPayload:
    base: Final = _make_standard_logging_payload_with_usage_object(usage_object={})
    return cast(
        StandardLoggingPayload,
        {
            **base,
            "status": "failure",
            "call_type": "aresponses",
            "model_id": "mid-123",
            "model_group": "group-x",
            "api_base": "https://api.openai.com/v1/responses",
            "custom_llm_provider": "openai",
        },
    )


def test_get_logging_payload_failed_request_falls_back_to_standard_logging_payload():
    """Failed-request kwargs from the proxy failure hook carry no deployment info
    (LIT-5795), so the attribution columns must come from the failure-time
    standard_logging_object."""
    payload = get_logging_payload(
        kwargs={
            "model": "group-x",
            "litellm_params": {"metadata": {"user_api_key": "test-key", "status": "failure"}},
            "standard_logging_object": _make_failed_request_standard_logging_payload(),
        },
        response_obj={},
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    assert payload["model_id"] == "mid-123"
    assert payload["model_group"] == "group-x"
    assert payload["api_base"] == "https://api.openai.com/v1/responses"
    assert payload["custom_llm_provider"] == "openai"


def test_get_logging_payload_request_kwargs_win_over_standard_logging_payload():
    payload = get_logging_payload(
        kwargs={
            "model": "group-y",
            "custom_llm_provider": "anthropic",
            "litellm_params": {
                "api_base": "https://kwargs.example.com",
                "metadata": {
                    "user_api_key": "test-key",
                    "model_group": "kwargs-group",
                    "model_info": {"id": "kwargs-mid"},
                },
            },
            "standard_logging_object": _make_failed_request_standard_logging_payload(),
        },
        response_obj={},
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    assert payload["model_id"] == "kwargs-mid"
    assert payload["model_group"] == "kwargs-group"
    assert payload["api_base"] == "https://kwargs.example.com"
    assert payload["custom_llm_provider"] == "anthropic"


def test_get_logging_payload_failed_request_without_standard_logging_payload_leaves_fields_empty():
    payload = get_logging_payload(
        kwargs={
            "model": "group-x",
            "litellm_params": {"metadata": {"user_api_key": "test-key", "status": "failure"}},
        },
        response_obj={},
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    assert payload["model_id"] == ""
    assert payload["model_group"] == ""
    assert payload["api_base"] == ""
    assert payload["custom_llm_provider"] == ""


class _ModelRouterSpendLogKwargs(TypedDict):
    model: ReadOnly[str]
    litellm_params: ReadOnly[dict[str, dict[str, str]]]
    standard_logging_object: ReadOnly[StandardLoggingPayload]


def _model_router_spend_log_kwargs(slp_model: str | None) -> _ModelRouterSpendLogKwargs:
    standard_logging_payload: Final = cast(
        StandardLoggingPayload,
        {
            "model": slp_model,
            "metadata": {},
            "model_map_information": StandardLoggingModelInformation(
                model_map_key="azure_ai/model_router", model_map_value=None
            ),
        },
    )
    return {
        "model": "azure_ai/model_router/model-router",
        "litellm_params": {"metadata": {"user_api_key": "sk-test-key"}},
        "standard_logging_object": standard_logging_payload,
    }


def test_get_logging_payload_uses_standard_logging_payload_model():
    payload = get_logging_payload(
        kwargs=_model_router_spend_log_kwargs(slp_model="azure_ai/gpt-5-mini"),
        response_obj={},
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    assert payload["model"] == "azure_ai/gpt-5-mini"


def test_get_logging_payload_falls_back_to_kwargs_model_when_slp_model_missing():
    payload = get_logging_payload(
        kwargs=_model_router_spend_log_kwargs(slp_model=None),
        response_obj={},
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    assert payload["model"] == "azure_ai/model_router/model-router"


@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_empty_key_slp_none_is_empty_string_not_none_literal():
    kwargs = {
        "model": "openai/gpt-4.1",
        "messages": [{"role": "user", "content": "Hello"}],
        "call_type": "acompletion",
        "litellm_params": {
            "metadata": {
                "user_api_key_user_id": "test_user",
            }
        },
    }
    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=Exception("error"),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )

    assert payload["api_key"] == "", (
        f"Expected empty string but got {payload['api_key']!r}; "
        "dropping _redact_logged_api_key's 'or \"\"' guard would yield 'None' here"
    )


def test_get_spend_logs_metadata_sibling_fields_preserved():
    raw = "anthropic-raw-key-xyz"
    meta = _get_spend_logs_metadata(
        {
            "user_api_key": raw,
            "user_api_key_alias": "my-alias",
            "user_api_key_team_id": "team-123",
        }
    )
    assert meta["user_api_key"] == hash_token(raw)
    assert meta["user_api_key_alias"] == "my-alias"
    assert meta["user_api_key_team_id"] == "team-123"


def test_redact_logged_api_key_partial_sha256_is_hashed():
    partial_hex = "a" * 63
    result = _redact_logged_api_key(partial_hex)
    assert result is not None
    assert result != partial_hex
    assert len(result) == 64
    assert result == hash_token(partial_hex)


def test_redact_logged_api_key_bearer_already_hashed_passes_through_with_flag():
    already_hashed = hash_token("sk-some-key")
    assert len(already_hashed) == 64
    result = _redact_logged_api_key(f"Bearer {already_hashed}", already_redacted=True)
    assert result == already_hashed
    assert hash_token(already_hashed) != result


def test_redact_logged_api_key_bearer_sha256_without_flag_is_hashed():
    already_hashed = hash_token("sk-some-key")
    assert len(already_hashed) == 64
    result = _redact_logged_api_key(f"Bearer {already_hashed}")
    assert result is not None
    assert result != already_hashed
    assert result == hash_token(already_hashed)


def test_autorouter_savings_flow_from_logging_payload_into_spend_log_metadata():
    """The figure the logging path computed is what the spend writer reads back, so it
    is threaded from the StandardLoggingPayload like cost_breakdown, never re-derived."""
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
            "standard_logging_object": {"autorouter_savings": 0.42, "metadata": {}, "model_map_information": None},
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-ar-savings", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["autorouter_savings"] == 0.42


@pytest.mark.parametrize("bucket", ["metadata", "litellm_metadata"])
def test_caller_forged_autorouter_savings_is_discarded(bucket):
    """The raw request bucket is client-writable and _get_spend_logs_metadata projects
    every SpendLogsMetadata key from it, so the logging payload's value must overwrite
    unconditionally or a caller could report savings the router never produced."""
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {bucket: {"user_api_key": "test-key", "autorouter_savings": 999.0}},
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-forged-savings", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["autorouter_savings"] is None


def test_get_logging_payload_includes_fallback_info_in_spend_logs_metadata():
    """
    Test that fallback info (attempted_fallbacks, original_model_group) from metadata
    is included in the spend logs metadata JSON.
    """
    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
                "attempted_fallbacks": 2,
                "original_model_group": "azure-gpt-fallback",
            }
        },
        "standard_logging_object": StandardLoggingPayload(
            id="test-fallback-123",
            call_type="completion",
            stream=False,
            response_cost=0.001,
            status="success",
            total_tokens=100,
            prompt_tokens=50,
            completion_tokens=50,
            startTime=1234567890.0,
            endTime=1234567891.0,
            completionStartTime=None,
            model_map_information=StandardLoggingModelInformation(
                model_map_key="gpt-3.5-turbo", model_map_value=None
            ),
            model="gpt-3.5-turbo",
            model_id="model-123",
            model_group="openai",
            custom_llm_provider="openai",
            api_base="https://api.openai.com",
            metadata=StandardLoggingMetadata(
                user_api_key_hash="test_hash",
                user_api_key_alias=None,
                user_api_key_team_id=None,
                user_api_key_org_id=None,
                user_api_key_user_id=None,
                user_api_key_team_alias=None,
                spend_logs_metadata=None,
                requester_ip_address=None,
                requester_metadata=None,
                user_api_key_end_user_id=None,
            ),
            cache_hit=False,
            cache_key=None,
            saved_cache_cost=0.0,
            request_tags=[],
            end_user=None,
            requester_ip_address=None,
            messages=[],
            response={},
            error_str=None,
            model_parameters={},
            hidden_params=StandardLoggingHiddenParams(
                model_id="model-123",
                cache_key=None,
                api_base="https://api.openai.com",
                response_cost="0.001",
                litellm_overhead_time_ms=None,
                additional_headers=None,
                batch_models=None,
                litellm_model_name=None,
                usage_object=None,
            ),
        ),
    }

    response_obj = {
        "id": "test-response-retry",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    metadata = json.loads(payload["metadata"])

    assert (
        metadata.get("attempted_fallbacks") == 2
    ), f"Expected attempted_fallbacks=2, got {metadata.get('attempted_fallbacks')}"
    assert (
        metadata.get("original_model_group") == "azure-gpt-fallback"
    ), f"Expected original_model_group=azure-gpt-fallback, got {metadata.get('original_model_group')}"


def test_get_logging_payload_handles_missing_fallback_info_gracefully():
    """
    Test that fallback fields are None when not present in metadata (backward compatibility).
    """
    kwargs = {
        "model": "gpt-3.5-turbo",
        "litellm_params": {
            "metadata": {
                "user_api_key": "sk-test-key",
            }
        },
        "standard_logging_object": StandardLoggingPayload(
            id="test-no-fallback-456",
            call_type="completion",
            stream=False,
            response_cost=0.001,
            status="success",
            total_tokens=100,
            prompt_tokens=50,
            completion_tokens=50,
            startTime=1234567890.0,
            endTime=1234567891.0,
            completionStartTime=None,
            model_map_information=StandardLoggingModelInformation(
                model_map_key="gpt-3.5-turbo", model_map_value=None
            ),
            model="gpt-3.5-turbo",
            model_id="model-123",
            model_group="openai",
            custom_llm_provider="openai",
            api_base="https://api.openai.com",
            metadata=StandardLoggingMetadata(
                user_api_key_hash="test_hash",
                user_api_key_alias=None,
                user_api_key_team_id=None,
                user_api_key_org_id=None,
                user_api_key_user_id=None,
                user_api_key_team_alias=None,
                spend_logs_metadata=None,
                requester_ip_address=None,
                requester_metadata=None,
                user_api_key_end_user_id=None,
            ),
            cache_hit=False,
            cache_key=None,
            saved_cache_cost=0.0,
            request_tags=[],
            end_user=None,
            requester_ip_address=None,
            messages=[],
            response={},
            error_str=None,
            model_parameters={},
            hidden_params=StandardLoggingHiddenParams(
                model_id="model-123",
                cache_key=None,
                api_base="https://api.openai.com",
                response_cost="0.001",
                litellm_overhead_time_ms=None,
                additional_headers=None,
                batch_models=None,
                litellm_model_name=None,
                usage_object=None,
            ),
        ),
    }

    response_obj = {
        "id": "test-response-no-fallback",
        "choices": [{"message": {"content": "Hello!"}}],
        "usage": {
            "total_tokens": 100,
            "prompt_tokens": 50,
            "completion_tokens": 50,
        },
    }

    start_time = datetime.datetime.now(timezone.utc)
    end_time = datetime.datetime.now(timezone.utc)

    payload = get_logging_payload(
        kwargs=kwargs,
        response_obj=response_obj,
        start_time=start_time,
        end_time=end_time,
    )

    metadata = json.loads(payload["metadata"])

    assert (
        metadata.get("attempted_fallbacks") is None
    ), "attempted_fallbacks should be None when not provided"
    assert (
        metadata.get("original_model_group") is None
    ), "original_model_group should be None when not provided"
@pytest.mark.parametrize("bucket", ["metadata", "litellm_metadata"])
def test_injected_cache_breakpoints_survive_into_spend_log_metadata(bucket):
    """The injection marker only gates savings if it reaches the spend-log row.

    _get_spend_logs_metadata projects onto SpendLogsMetadata.__annotations__, so an
    undeclared key is dropped silently. Both buckets are covered because chat routes
    stamp metadata while /v1/messages routes stamp litellm_metadata, and
    record_gateway_injection writes into whichever the request carries.
    """
    payload = get_logging_payload(
        kwargs={
            "model": "claude-sonnet-5",
            "litellm_params": {
                bucket: {
                    "user_api_key": "test-key",
                    "litellm_gateway_injected_cache": "dep-of-this-row",
                }
            },
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-injected", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["litellm_gateway_injected_cache"] == "dep-of-this-row"


def test_passthrough_caching_carries_no_injection_marker():
    """The negative class the gate depends on: a request whose cache_control the client
    supplied must read as unmarked, not merely unlabelled by accident."""
    payload = get_logging_payload(
        kwargs={
            "model": "claude-sonnet-5",
            "litellm_params": {"metadata": {"user_api_key": "test-key"}},
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-passthrough", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["litellm_gateway_injected_cache"] is None


def _routed_call_kwargs(model_info: Mapping[str, object]) -> dict[str, object]:
    return {
        "model": "claude-haiku-4-5",
        "custom_llm_provider": "azure_ai",
        "litellm_call_id": "router-corr-123",
        "litellm_params": {
            "metadata": {
                "user_api_key": "test-key",
                "model_group": "internal-router/gpt-5.4",
                "deployment": "azure_ai/claude-haiku-4-5",
                "model_info": model_info,
            }
        },
    }


def test_router_metadata_stamped_for_internal_router_model_deployment():
    """A deployment flagged model_info.internal_router_model gets a router_metadata
    block correlating the requested model group with the selected deployment."""
    payload = get_logging_payload(
        kwargs=_routed_call_kwargs({"id": "mi-1", "internal_router_model": True}),
        response_obj=litellm.ModelResponse(id="chatcmpl-router-meta", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["router_metadata"] == {
        "requested_model": "internal-router/gpt-5.4",
        "selected_model": "azure_ai/claude-haiku-4-5",
        "selected_provider": "azure_ai",
        "router_correlation_id": "router-corr-123",
    }


def test_router_metadata_absent_without_internal_router_model_flag():
    payload = get_logging_payload(
        kwargs=_routed_call_kwargs({"id": "mi-1"}),
        response_obj=litellm.ModelResponse(id="chatcmpl-unflagged", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["router_metadata"] is None


@pytest.mark.parametrize("bucket", ["metadata", "litellm_metadata"])
def test_caller_forged_router_metadata_is_discarded(bucket):
    """The raw request bucket is client-writable and _get_spend_logs_metadata projects
    every SpendLogsMetadata key from it, so the server-derived value must overwrite
    unconditionally or a caller could plant router provenance the router never produced."""
    payload = get_logging_payload(
        kwargs={
            "model": "gpt-4o-mini",
            "litellm_params": {
                bucket: {
                    "user_api_key": "test-key",
                    "router_metadata": {"requested_model": "forged", "router_correlation_id": "forged-id"},
                }
            },
        },
        response_obj=litellm.ModelResponse(id="chatcmpl-forged-router-meta", choices=[], usage=litellm.Usage()),
        start_time=datetime.datetime.now(timezone.utc),
        end_time=datetime.datetime.now(timezone.utc),
    )
    metadata = json.loads(payload["metadata"])
    assert metadata["router_metadata"] is None
