Merge 74ebc06069 into 220a70653e

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -1324,7 +1324,7 @@ def get_thought_signature_from_tool(tool: Mapping[str, object]) -> str | None:
 _get_thought_signature_from_tool = get_thought_signature_from_tool
 
 
-def _get_dummy_thought_signature() -> str:
+def get_dummy_thought_signature() -> str:
     """Generate a dummy thought signature for models that require it.
 
     This is used when transferring conversation history from older models
@@ -1341,6 +1341,9 @@ def _get_dummy_thought_signature() -> str:
     return base64.b64encode(dummy_data).decode("utf-8")
 
 
+_get_dummy_thought_signature: Final = get_dummy_thought_signature
+
+
 def convert_to_gemini_tool_call_invoke(
     message: ChatCompletionAssistantMessage,
     model: str | None = None,
@@ -1411,7 +1414,7 @@ def convert_to_gemini_tool_call_invoke(
                         # https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/thinking/thought-signatures#parallel_function_calling_example
                         is_first_function_call = len(_parts_list) == 0
                         if not thought_signature and is_first_function_call and needs_dummy_signature:
-                            thought_signature = _get_dummy_thought_signature()
+                            thought_signature = get_dummy_thought_signature()
                         if thought_signature:
                             part_dict["thoughtSignature"] = thought_signature
 
@@ -1435,7 +1438,7 @@ def convert_to_gemini_tool_call_invoke(
 
                 # If no signature found and model is gemini-3, use dummy signature
                 if not thought_signature and needs_dummy_signature:
-                    thought_signature = _get_dummy_thought_signature()
+                    thought_signature = get_dummy_thought_signature()
 
                 if thought_signature:
                     part_dict_function["thoughtSignature"] = thought_signature

litellm/llms/vertex_ai/context_caching/rejected_cache_retry.py

@@ -26,7 +26,7 @@ def cache_rebuild(
     return partial(build, messages=deepcopy(messages), optional_params=independent_snapshot(optional_params))
 
 
-def _status_and_message(error: httpx.HTTPStatusError | BaseLLMException) -> tuple[int, str]:
+def status_and_message(error: httpx.HTTPStatusError | BaseLLMException) -> tuple[int, str]:
     match error:
         case httpx.HTTPStatusError():
             return error.response.status_code, error.response.text
@@ -35,7 +35,7 @@ def _status_and_message(error: httpx.HTTPStatusError | BaseLLMException) -> tupl
 
 
 def _cache_rejected(body: RequestBody, error: httpx.HTTPStatusError | BaseLLMException) -> bool:
-    return body.get("cachedContent") is not None and names_rejected_cache(*_status_and_message(error))
+    return body.get("cachedContent") is not None and names_rejected_cache(*status_and_message(error))
 
 
 async def post_with_fresh_cache_async(

litellm/llms/vertex_ai/gemini/rejected_thought_signature_retry.py

@@ -0,0 +1,92 @@
+from collections.abc import Awaitable, Callable, Iterator
+from typing import Final, TypeVar
+
+import httpx
+
+from litellm import verbose_logger
+from litellm.litellm_core_utils.prompt_templates.factory import get_dummy_thought_signature
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.types.llms.vertex_ai import ContentType, PartType, RequestBody
+
+from ..context_caching.rejected_cache_retry import status_and_message
+
+Sent: Final = TypeVar("Sent")
+
+_REJECTED_SIGNATURE_MARKER: Final = "thought signature"
+
+
+def names_rejected_thought_signature(status_code: int, message: str) -> bool:
+    """A signature only validates on the backend that minted it. AI Studio answers a replayed Vertex signature
+    with ``400 Corrupted thought signature.`` and Vertex answers an AI Studio one with ``400 Invalid thought
+    signature.``"""
+    return status_code == 400 and _REJECTED_SIGNATURE_MARKER in message.lower()
+
+
+def _foreign_signatures(body: RequestBody, placeholder: str) -> Iterator[str]:
+    for content in body["contents"]:
+        for part in content["parts"]:
+            if (signature := part.get("thoughtSignature")) and signature != placeholder:
+                yield signature
+
+
+def _part_with_placeholder(part: PartType, placeholder: str) -> PartType:
+    if not part.get("thoughtSignature"):
+        return part
+    return {**part, "thoughtSignature": placeholder}
+
+
+def _content_with_placeholders(content: ContentType, placeholder: str) -> ContentType:
+    return {**content, "parts": [_part_with_placeholder(part, placeholder) for part in content["parts"]]}
+
+
+def with_placeholder_thought_signatures(body: RequestBody) -> RequestBody | None:
+    """Swaps every thought signature in the history for Google's documented skip-validator placeholder, or
+    returns ``None`` when the body carries nothing the backend could have rejected."""
+    placeholder: Final = get_dummy_thought_signature()
+    if next(_foreign_signatures(body, placeholder), None) is None:
+        return None
+    return {**body, "contents": [_content_with_placeholders(content, placeholder) for content in body["contents"]]}
+
+
+def _resend_body(body: RequestBody, error: httpx.HTTPStatusError | BaseLLMException) -> RequestBody | None:
+    if not names_rejected_thought_signature(*status_and_message(error)):
+        return None
+    resend_body: Final = with_placeholder_thought_signatures(body)
+    if resend_body is not None:
+        verbose_logger.warning(
+            "Gemini rejected a replayed thought signature, resending once with the skip_thought_signature_validator"
+            " placeholder in every signature slot"
+        )
+    return resend_body
+
+
+async def post_with_accepted_thought_signatures_async(
+    body: RequestBody,
+    post: Callable[[RequestBody], Awaitable[Sent]],
+    log_request: Callable[[RequestBody], None],
+) -> tuple[RequestBody, Sent]:
+    """Innermost retry around the raw post, so a cache rebuild (which re-transforms the messages and brings
+    the rejected signatures back) is itself wrapped by it. Returns the body the backend accepted."""
+    try:
+        return body, await post(body)
+    except (httpx.HTTPStatusError, BaseLLMException) as error:
+        resend_body: Final = _resend_body(body, error)
+        if resend_body is None:
+            raise
+        log_request(resend_body)
+    return resend_body, await post(resend_body)
+
+
+def post_with_accepted_thought_signatures_sync(
+    body: RequestBody,
+    post: Callable[[RequestBody], Sent],
+    log_request: Callable[[RequestBody], None],
+) -> tuple[RequestBody, Sent]:
+    try:
+        return body, post(body)
+    except (httpx.HTTPStatusError, BaseLLMException) as error:
+        resend_body: Final = _resend_body(body, error)
+        if resend_body is None:
+            raise
+        log_request(resend_body)
+    return resend_body, post(resend_body)

litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py

@@ -133,6 +133,10 @@ from .long_decode import (
     with_initial_prompt_tokens,
     with_spans_shifted,
 )
+from .rejected_thought_signature_retry import (
+    post_with_accepted_thought_signatures_async,
+    post_with_accepted_thought_signatures_sync,
+)
 from .transformation import (
     async_transform_request_body,
     gemini_convert_messages_with_history,
@@ -3472,7 +3476,8 @@ class VertexLLM(VertexBase):
             async def post(body: RequestBody) -> ModelResponseIterator:
                 return await open_stream(client=client, data=json.dumps(body), request_body=body)
 
-            return (await post_with_fresh_cache_async(request_body, rebuild, post, log_request))[1]
+            accepted: Final = partial(post_with_accepted_thought_signatures_async, post=post, log_request=log_request)
+            return (await post_with_fresh_cache_async(request_body, rebuild, accepted, log_request))[1][1]
 
         streaming_response: Final = CustomStreamWrapper(
             completion_stream=None,
@@ -3596,7 +3601,8 @@ class VertexLLM(VertexBase):
                 )
             return response
 
-        sent_body, first_response = await post_with_fresh_cache_async(request_body, rebuild, post, log_request)
+        accepted: Final = partial(post_with_accepted_thought_signatures_async, post=post, log_request=log_request)
+        _, (sent_body, first_response) = await post_with_fresh_cache_async(request_body, rebuild, accepted, log_request)
         return _transform_long_decode_hops(
             model=model,
             hops=await collect_hops_async(parse_body(first_response), sent_body, post),
@@ -3772,7 +3778,10 @@ class VertexLLM(VertexBase):
                 def post(body: RequestBody) -> ModelResponseIterator:
                     return open_stream(client=client, data=json.dumps(body), request_body=body)
 
-                return post_with_fresh_cache_sync(data, rebuild, post, log_request)[1]
+                accepted: Final = partial(
+                    post_with_accepted_thought_signatures_sync, post=post, log_request=log_request
+                )
+                return post_with_fresh_cache_sync(data, rebuild, accepted, log_request)[1][1]
 
             streaming_response: Final = CustomStreamWrapper(
                 completion_stream=None,
@@ -3816,7 +3825,8 @@ class VertexLLM(VertexBase):
                 )
             return response
 
-        sent_body, first_response = post_with_fresh_cache_sync(data, rebuild, post, log_request)
+        accepted: Final = partial(post_with_accepted_thought_signatures_sync, post=post, log_request=log_request)
+        _, (sent_body, first_response) = post_with_fresh_cache_sync(data, rebuild, accepted, log_request)
         return _transform_long_decode_hops(
             model=model,
             hops=collect_hops_sync(parse_body(first_response), sent_body, post),

tests/integration/providers/test_gemini_thinking_replay_wire.py

@@ -2,6 +2,7 @@ import base64
 import itertools
 import json
 import os
+import re
 import uuid
 from collections.abc import Iterator, Mapping, Sequence
 from collections.abc import Set as AbstractSet
@@ -679,9 +680,70 @@ def test_gemini_own_tool_call_signature_is_still_replayed_on_the_function_call_p
         assert _spend_row(rounds[1])["status"] == "success"
 
 
-def test_responses_previous_response_id_replays_gemini_session_history(
-    gateway: Gateway, tmp_path: Path
+_PLACEHOLDER_SIGNATURE: Final = base64.b64encode(b"skip_thought_signature_validator").decode()
+_ISSUED_ID: Final = re.compile(r'"id":\s*"([^"]+)"')
+_PLACEHOLDER_TOOL_TURN: Final = _model_turn(
+    _thought("I should count."),
+    {"function_call": {"name": "count_fruit", "args": {"kind": "apple"}}, "thoughtSignature": _PLACEHOLDER_SIGNATURE},
+)
+# Wording AI Studio answers with when a Vertex-minted signature is replayed to it, captured live on 2026-10-27
+_FOREIGN_SIGNATURE_REJECTED: Final = Reply(
+    status=400,
+    body=json.dumps(
+        {"error": {"code": 400, "message": "Corrupted thought signature.", "status": "INVALID_ARGUMENT"}}
+    ).encode(),
+)
+
+
+@pytest.mark.parametrize("stream", (False, True))
+@pytest.mark.parametrize("endpoint", _ENDPOINTS)
+def test_gemini_tool_call_signature_the_backend_rejects_is_resent_once_with_the_placeholder(
+    gateway: Gateway, endpoint: Endpoint, stream: bool
 ) -> None:
+    rounds: Final = (f"gemini-reply-{uuid.uuid4().hex}", f"gemini-reply-{uuid.uuid4().hex}")
+    calls: Final = itertools.count()
+
+    def respond(request: Request) -> Reply:
+        match next(calls):
+            case 0:
+                return _reply(rounds[0], stream=False, parts=_GEMINI_TOOL_REPLY_PARTS)
+            case 1:
+                return _FOREIGN_SIGNATURE_REJECTED
+            case _:
+                return _reply(rounds[1], stream=stream, parts=({"text": "Twelve apples."},))
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = _register(gateway, scenario, "gemini", wire)
+        first: Final = _tool_round_one(endpoint, model)
+        headers: Final = {"anthropic-version": "2023-06-01"}
+        opened: Final = gateway.request("POST", _path(endpoint), first, headers=headers)
+        assert opened.status_code == 200, opened.text
+        second: Final = {
+            **_tool_round_two(endpoint, first, _JSON_OBJECT.validate_json(opened.content)),
+            "stream": stream,
+        }
+        closed: Final = gateway.request("POST", _path(endpoint), second, headers=headers)
+        assert closed.status_code == 200, closed.text
+        assert "Twelve apples." in closed.text, closed.text
+        received: Final = wire.drain()
+        assert [(request.method, request.target) for request in received] == [
+            ("POST", _target("gemini", False)),
+            ("POST", _target("gemini", stream)),
+            ("POST", _target("gemini", stream)),
+        ], received
+        rejected, resent = (_JSON_OBJECT.validate_json(request.body) for request in received[1:])
+        assert rejected == {
+            **_expected_body(endpoint, _user("Count the apples."), _GEMINI_TOOL_REPLAYED_TURN, _FUNCTION_RESPONSE_TURN),
+            **_DECLARATIONS,
+        }, rejected
+        assert resent == {
+            **_expected_body(endpoint, _user("Count the apples."), _PLACEHOLDER_TOOL_TURN, _FUNCTION_RESPONSE_TURN),
+            **_DECLARATIONS,
+        }, resent
+        assert _spend_row(rounds[1], *_ISSUED_ID.findall(closed.text))["status"] == "success"
+
+
+def test_responses_previous_response_id_replays_gemini_session_history(gateway: Gateway, tmp_path: Path) -> None:
     first_prompt: Final = f"gemini-session-first-{uuid.uuid4().hex}"
     second_prompt: Final = f"gemini-session-second-{uuid.uuid4().hex}"
     first_answer: Final = f"gemini-answer-first-{uuid.uuid4().hex}"

tests/unit/llms/vertex_ai/gemini/test_rejected_thought_signature_retry.py

@@ -0,0 +1,228 @@
+import base64
+import json
+from collections.abc import Callable, Mapping, Sequence
+from typing import Final
+
+import httpx
+import pytest
+
+import litellm
+from litellm.litellm_core_utils.prompt_templates.factory import get_dummy_thought_signature
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+from litellm.llms.vertex_ai.gemini.rejected_thought_signature_retry import with_placeholder_thought_signatures
+from litellm.types.llms.vertex_ai import RequestBody
+from litellm.types.utils import Choices, ModelResponse, ModelResponseStream
+from litellm.utils import CustomStreamWrapper
+
+MODEL: Final = "gemini/gemini-3.8-flash"
+VERTEX_MINTED: Final = base64.b64encode(b"minted-by-vertex").decode()
+VERTEX_MINTED_TEXT: Final = base64.b64encode(b"minted-by-vertex-for-text").decode()
+HISTORY_WITH_FOREIGN_SIGNATURES: Final = (
+    {"role": "user", "content": "What is the weather in Paris?"},
+    {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [
+            {
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'},
+                "provider_specific_fields": {"thought_signature": VERTEX_MINTED},
+            }
+        ],
+    },
+    {"role": "tool", "tool_call_id": "call_1", "content": "18C and sunny"},
+    {
+        "role": "assistant",
+        "content": "It is 18C and sunny in Paris.",
+        "provider_specific_fields": {"thought_signatures": [VERTEX_MINTED_TEXT]},
+    },
+    {"role": "user", "content": "And in Berlin?"},
+)
+HISTORY_WITHOUT_SIGNATURES: Final = (
+    {"role": "user", "content": "What is the weather in Paris?"},
+    {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [
+            {"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}}
+        ],
+    },
+    {"role": "tool", "tool_call_id": "call_1", "content": "18C and sunny"},
+    {"role": "user", "content": "And in Berlin?"},
+)
+TOOLS: Final = (
+    {
+        "type": "function",
+        "function": {
+            "name": "get_weather",
+            "parameters": {"type": "object", "properties": {"city": {"type": "string"}}},
+        },
+    },
+)
+ANSWER: Final = {
+    "candidates": [{"content": {"role": "model", "parts": [{"text": "Berlin is 12C."}]}, "finishReason": "STOP"}],
+    "usageMetadata": {"promptTokenCount": 40, "candidatesTokenCount": 6, "totalTokenCount": 46},
+}
+# Wording observed from Google on 2026-10-08 when a signature minted by the other backend is replayed
+AI_STUDIO_REJECTION: Final = "Corrupted thought signature."
+VERTEX_REJECTION: Final = "Invalid thought signature."
+UNRELATED_REJECTION: Final = 'Invalid JSON payload received. Unknown name "foo".'
+
+
+def _rejected(message: str) -> httpx.Response:
+    return httpx.Response(400, json={"error": {"code": 400, "message": message, "status": "INVALID_ARGUMENT"}})
+
+
+def _sse(*chunks: Mapping[str, object]) -> httpx.Response:
+    return httpx.Response(
+        200,
+        content="".join(f"data: {json.dumps(chunk)}\r\n\r\n" for chunk in chunks).encode(),
+        headers={"content-type": "text/event-stream"},
+    )
+
+
+def _replay(
+    answers: Sequence[httpx.Response], sent: list[dict[str, object]]
+) -> Callable[[httpx.Request], httpx.Response]:
+    def handler(request: httpx.Request) -> httpx.Response:
+        sent.append(json.loads(request.content))
+        return answers[len(sent) - 1]
+
+    return handler
+
+
+def _signatures_of(body: Mapping[str, object]) -> list[str]:
+    contents: Final = body["contents"]
+    assert isinstance(contents, list)
+    parts: Final = [part for content in contents for part in content["parts"]]
+    return [part["thoughtSignature"] for part in parts if "thoughtSignature" in part]
+
+
+def _text_of(response: ModelResponse) -> str:
+    choice: Final = response.choices[0]
+    assert isinstance(choice, Choices), choice
+    assert isinstance(choice.message.content, str), choice.message
+    return choice.message.content
+
+
+def _delta_text_of(chunk: object) -> str:
+    assert isinstance(chunk, ModelResponseStream), chunk
+    return chunk.choices[0].delta.content or ""
+
+
+async def _complete(
+    messages: Sequence[Mapping[str, object]],
+    answers: Sequence[httpx.Response],
+    sent: list[dict[str, object]],
+    is_async: bool,
+    stream: bool,
+) -> str:
+    kwargs: Final = {
+        "model": MODEL,
+        "messages": list(messages),
+        "tools": list(TOOLS),
+        "stream": stream,
+        "api_key": "unit-test-key",
+    }
+    if is_async:
+        response = await litellm.acompletion(
+            **kwargs, client=AsyncHTTPHandler(transport=httpx.MockTransport(_replay(answers, sent)))
+        )
+    else:
+        response = litellm.completion(
+            **kwargs, client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(_replay(answers, sent))))
+        )
+    if isinstance(response, ModelResponse):
+        return _text_of(response)
+    assert isinstance(response, CustomStreamWrapper), response
+    if is_async:
+        return "".join([_delta_text_of(chunk) async for chunk in response])
+    return "".join(_delta_text_of(chunk) for chunk in response)
+
+
+def _answer(stream: bool) -> httpx.Response:
+    return _sse(ANSWER) if stream else httpx.Response(200, json=ANSWER)
+
+
+@pytest.mark.parametrize("is_async", [True, False])
+@pytest.mark.parametrize("stream", [True, False])
+@pytest.mark.parametrize("rejection", [AI_STUDIO_REJECTION, VERTEX_REJECTION])
+@pytest.mark.asyncio
+async def test_signature_rejected_by_the_other_backend_is_resent_once_with_the_placeholder(
+    is_async: bool, stream: bool, rejection: str
+):
+    sent: Final[list[dict[str, object]]] = []
+    answer: Final = await _complete(
+        HISTORY_WITH_FOREIGN_SIGNATURES, (_rejected(rejection), _answer(stream)), sent, is_async, stream
+    )
+    assert answer == "Berlin is 12C."
+    assert len(sent) == 2
+    first_signatures, second_signatures = _signatures_of(sent[0]), _signatures_of(sent[1])
+    assert set(first_signatures) == {VERTEX_MINTED, VERTEX_MINTED_TEXT}
+    assert second_signatures == [get_dummy_thought_signature()] * len(first_signatures)
+    assert {**sent[1], "contents": None} == {**sent[0], "contents": None}
+
+
+@pytest.mark.parametrize("is_async", [True, False])
+@pytest.mark.parametrize("stream", [True, False])
+@pytest.mark.asyncio
+async def test_placeholder_rejected_too_is_raised_after_the_second_post(is_async: bool, stream: bool):
+    sent: Final[list[dict[str, object]]] = []
+    with pytest.raises(litellm.BadRequestError, match="thought signature"):
+        await _complete(
+            HISTORY_WITH_FOREIGN_SIGNATURES,
+            (_rejected(AI_STUDIO_REJECTION), _rejected(AI_STUDIO_REJECTION)),
+            sent,
+            is_async,
+            stream,
+        )
+    assert len(sent) == 2
+
+
+@pytest.mark.parametrize("is_async", [True, False])
+@pytest.mark.parametrize("stream", [True, False])
+@pytest.mark.asyncio
+async def test_unrelated_400_is_raised_without_a_resend(is_async: bool, stream: bool):
+    sent: Final[list[dict[str, object]]] = []
+    with pytest.raises(litellm.BadRequestError, match="Unknown name"):
+        await _complete(HISTORY_WITH_FOREIGN_SIGNATURES, (_rejected(UNRELATED_REJECTION),), sent, is_async, stream)
+    assert len(sent) == 1
+
+
+@pytest.mark.parametrize("is_async", [True, False])
+@pytest.mark.parametrize("stream", [True, False])
+@pytest.mark.asyncio
+async def test_signature_400_on_a_history_carrying_only_the_placeholder_is_raised_without_a_resend(
+    is_async: bool, stream: bool
+):
+    sent: Final[list[dict[str, object]]] = []
+    with pytest.raises(litellm.BadRequestError, match="thought signature"):
+        await _complete(HISTORY_WITHOUT_SIGNATURES, (_rejected(AI_STUDIO_REJECTION),), sent, is_async, stream)
+    assert len(sent) == 1
+    assert set(_signatures_of(sent[0])) <= {get_dummy_thought_signature()}
+
+
+def test_placeholder_body_keeps_the_original_untouched():
+    body: Final[RequestBody] = {
+        "contents": [
+            {"role": "model", "parts": [{"text": "hi", "thoughtSignature": VERTEX_MINTED}, {"text": "unsigned"}]},
+            {"role": "user", "parts": [{"text": "again"}]},
+        ],
+        "generationConfig": {"temperature": 0.1},
+    }
+    resend: Final = with_placeholder_thought_signatures(body)
+    assert resend is not None
+    assert _signatures_of(resend) == [get_dummy_thought_signature()]
+    assert _signatures_of(body) == [VERTEX_MINTED]
+    assert resend.get("generationConfig") == body.get("generationConfig")
+
+
+def test_body_without_a_foreign_signature_has_nothing_to_resend():
+    body: Final[RequestBody] = {
+        "contents": [
+            {"role": "model", "parts": [{"text": "hi", "thoughtSignature": get_dummy_thought_signature()}]},
+            {"role": "user", "parts": [{"text": "again"}]},
+        ]
+    }
+    assert with_placeholder_thought_signatures(body) is None

tests/unit/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py

@@ -2824,7 +2824,7 @@ def test_gemini_25_text_signature_survives_replay_to_gemini_3():
     """gemini-2.5 history (signed text, unsigned tool call) replayed to gemini-3 keeps the real
     text signature; the dummy signature synthesized for the unsigned tool call must not suppress it."""
     from litellm.litellm_core_utils.prompt_templates.factory import (
-        _get_dummy_thought_signature,
+        get_dummy_thought_signature,
     )
     from litellm.llms.vertex_ai.gemini.transformation import (
         gemini_convert_messages_with_history,
@@ -2848,7 +2848,7 @@ def test_gemini_25_text_signature_survives_replay_to_gemini_3():
     assert parts[0]["text"] == "I will list the directory."
     assert parts[0]["thoughtSignature"] == "real_25_signature"
     assert "function_call" in parts[1]
-    assert parts[1]["thoughtSignature"] == _get_dummy_thought_signature()
+    assert parts[1]["thoughtSignature"] == get_dummy_thought_signature()
 
 
 def test_gemini_function_call_signature_round_trip_no_duplicate():
@@ -2990,7 +2990,7 @@ def test_thinking_block_signature_is_not_forwarded_to_gemini() -> None:
 
 
 def test_anthropic_messages_history_replays_to_gemini_without_claude_signature() -> None:
-    from litellm.litellm_core_utils.prompt_templates.factory import _get_dummy_thought_signature
+    from litellm.litellm_core_utils.prompt_templates.factory import get_dummy_thought_signature
     from litellm.llms.anthropic.messages.adapters.transformation import LiteLLMAnthropicMessagesAdapter
 
     thinking: Final = "I should look the weather up before answering."
@@ -3013,6 +3013,6 @@ def test_anthropic_messages_history_replays_to_gemini_without_claude_signature()
     parts: Final = _parts_of(gemini_convert_messages_with_history(messages=chat_messages, model="gemini-3.8-flash"))
 
     signatures: Final = [part["thoughtSignature"] for part in parts if "thoughtSignature" in part]
-    assert signatures == [_get_dummy_thought_signature()]
+    assert signatures == [get_dummy_thought_signature()]
     assert [part for part in parts if part.get("text") == thinking] == [{"thought": True, "text": thinking}]
-    assert next(part for part in parts if "function_call" in part)["thoughtSignature"] == _get_dummy_thought_signature()
+    assert next(part for part in parts if "function_call" in part)["thoughtSignature"] == get_dummy_thought_signature()