diff --git a/litellm/litellm_core_utils/prompt_templates/factory.py b/litellm/litellm_core_utils/prompt_templates/factory.py index fad4caa59a0..44b54f89b60 100644 --- a/litellm/litellm_core_utils/prompt_templates/factory.py +++ b/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 diff --git a/litellm/llms/vertex_ai/context_caching/rejected_cache_retry.py b/litellm/llms/vertex_ai/context_caching/rejected_cache_retry.py index 43e4ee8dc84..3b0f210df7c 100644 --- a/litellm/llms/vertex_ai/context_caching/rejected_cache_retry.py +++ b/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( diff --git a/litellm/llms/vertex_ai/gemini/rejected_thought_signature_retry.py b/litellm/llms/vertex_ai/gemini/rejected_thought_signature_retry.py new file mode 100644 index 00000000000..ae51efb2b34 --- /dev/null +++ b/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) diff --git a/litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py b/litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py index 9acbc9c459c..53d412c7300 100644 --- a/litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py +++ b/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), diff --git a/tests/integration/providers/test_gemini_thinking_replay_wire.py b/tests/integration/providers/test_gemini_thinking_replay_wire.py index 4f965f079ec..70e339ed8b7 100644 --- a/tests/integration/providers/test_gemini_thinking_replay_wire.py +++ b/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}" diff --git a/tests/unit/llms/vertex_ai/gemini/test_rejected_thought_signature_retry.py b/tests/unit/llms/vertex_ai/gemini/test_rejected_thought_signature_retry.py new file mode 100644 index 00000000000..04312fb0dc1 --- /dev/null +++ b/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 diff --git a/tests/unit/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py b/tests/unit/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py index c0ff931a41d..2c8c7520c38 100644 --- a/tests/unit/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py +++ b/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()