fix(google): add fallback thought signature sentinel for foreign function calls

livekit-agents/livekit/agents/llm/_provider_format/google.py

@@ -10,6 +10,12 @@
 
 from .utils import convert_mid_conversation_instructions, group_tool_calls
 
+# Google's documented sentinel for function calls that were not generated by the
+# Gemini API and therefore have no thought signature (e.g. history transferred
+# from another provider through FallbackAdapter).
+# https://ai.google.dev/gemini-api/docs/thought-signatures
+_SKIP_THOUGHT_SIGNATURE_SENTINEL = b"skip_thought_signature_validator"
+
 
 @dataclass
 class GoogleFormatData:
@@ -68,9 +74,14 @@ def to_chat_ctx(
                     "args": json.loads(msg.arguments or "{}"),
                 }
             }
-            # Inject thought_signature if available (Gemini 3 multi-turn function calling)
-            if thought_signatures and (sig := thought_signatures.get(msg.call_id)):
-                fc_part["thought_signature"] = sig
+            # Gemini 2.5+ requires a thought_signature on every function_call part in
+            # multi-turn conversations. A call_id without a stored signature means the
+            # call came from another provider (e.g. via FallbackAdapter), so fall back
+            # to Google's validator-skip sentinel instead of omitting the field.
+            if thought_signatures is not None:
+                fc_part["thought_signature"] = thought_signatures.get(
+                    msg.call_id, _SKIP_THOUGHT_SIGNATURE_SENTINEL
+                )
             parts.append(fc_part)
         elif msg.type == "function_call_output":
             response = {"output": msg.output} if not msg.is_error else {"error": msg.output}

livekit-plugins/livekit-plugins-google/livekit/plugins/google/llm.py

@@ -441,10 +441,12 @@ async def _run(self) -> None:
         request_id = utils.shortuuid()
 
         try:
-            # Pass thought_signatures for Gemini 2.5+ multi-turn function calling
-            thought_sigs = (
-                self._llm._thought_signatures if _requires_thought_signatures(self._model) else None
-            )
+            # Pass thought_signatures for Gemini 2.5+ multi-turn function calling.
+            # The cache may be None on a fresh instance; pass an empty mapping so
+            # the formatter can still inject the fallback sentinel.
+            thought_sigs: dict[str, bytes] | None = None
+            if _requires_thought_signatures(self._model):
+                thought_sigs = getattr(self._llm, "_thought_signatures", None) or {}
             turns_dict, extra_data = self._chat_ctx.to_provider_format(
                 format="google", thought_signatures=thought_sigs
             )
@@ -614,6 +616,8 @@ def _parse_part(self, id: str, part: types.Part) -> llm.ChatChunk | None:
                 and hasattr(part, "thought_signature")
                 and part.thought_signature
             ):
+                if getattr(self._llm, "_thought_signatures", None) is None:
+                    self._llm._thought_signatures = {}
                 self._llm._thought_signatures[tool_call.call_id] = part.thought_signature
 
             chat_chunk = llm.ChatChunk(

tests/test_google_thought_signatures.py

@@ -1,5 +1,6 @@
 import pytest
 
+from livekit.agents.llm import ChatContext, FunctionCall, FunctionCallOutput
 from livekit.plugins.google.llm import (
     _is_gemini_3_flash_model,
     _is_gemini_3_model,
@@ -9,6 +10,42 @@
 pytestmark = pytest.mark.unit
 
 
+def _ctx_with_function_call(call_id: str) -> ChatContext:
+    ctx = ChatContext.empty()
+    ctx.add_message(role="user", content="hello")
+    ctx.insert(FunctionCall(call_id=call_id, name="tool", arguments="{}"))
+    ctx.insert(FunctionCallOutput(call_id=call_id, name="tool", output="ok", is_error=False))
+    return ctx
+
+
+class TestThoughtSignatureFormatting:
+    """Formatting of thought_signature on function_call parts (see #6135)."""
+
+    def test_stored_signature_is_passed_through(self):
+        ctx = _ctx_with_function_call("call_1")
+
+        turns, _ = ctx.to_provider_format(
+            format="google", thought_signatures={"call_1": b"real_signature"}
+        )
+
+        assert turns[1]["parts"][0]["thought_signature"] == b"real_signature"
+
+    def test_unknown_call_id_gets_skip_sentinel(self):
+        ctx = _ctx_with_function_call("call_from_another_provider")
+
+        turns, _ = ctx.to_provider_format(format="google", thought_signatures={})
+
+        part = turns[1]["parts"][0]
+        assert part["thought_signature"] == b"skip_thought_signature_validator"
+
+    def test_no_signature_mapping_omits_field(self):
+        ctx = _ctx_with_function_call("call_1")
+
+        turns, _ = ctx.to_provider_format(format="google")
+
+        assert "thought_signature" not in turns[1]["parts"][0]
+
+
 class TestGeminiModelDetection:
     """Tests for Gemini model detection helper functions."""
 

tests/test_plugin_google_llm.py

@@ -6,7 +6,7 @@
 from google.genai import types
 
 from livekit.agents import llm
-from livekit.agents.llm import ChatContext, function_tool
+from livekit.agents.llm import ChatContext, FunctionCall, FunctionCallOutput, function_tool
 from livekit.agents.types import APIConnectOptions
 from livekit.plugins.google.llm import LLM, LLMStream
 from livekit.plugins.google.realtime.realtime_api import RealtimeModel, RealtimeSession
@@ -70,6 +70,21 @@ def test_empty_text_part_returns_none(self, llm_stream: LLMStream):
 
         assert chunk is None
 
+    def test_signature_stored_even_when_cache_is_none(self, llm_stream: LLMStream):
+        llm_stream._model = "gemini-2.5-flash"
+        llm_stream._llm._thought_signatures = None
+        part = types.Part(
+            function_call=types.FunctionCall(
+                id="call_1", name="get_weather", args={"city": "Paris"}
+            ),
+            thought_signature=b"real_signature",
+        )
+
+        chunk = llm_stream._parse_part("test-id", part)
+
+        assert chunk is not None
+        assert llm_stream._llm._thought_signatures == {"call_1": b"real_signature"}
+
 
 class TestCachedContentOption:
     """Verify the ``cached_content`` constructor option propagates from
@@ -273,6 +288,53 @@ async def test_request_merges_timeout_into_caller_http_options(self) -> None:
         assert caller_http_options.headers == {"X-Vertex-Test": "1"}
 
 
+class TestThoughtSignatureRequests:
+    """Request-level thought_signature behaviour (issue #6135): Gemini 2.5+
+    must receive Google's validator-skip sentinel for function calls made by
+    another provider (e.g. through FallbackAdapter), while pre-2.5 models must
+    never be sent the field."""
+
+    @staticmethod
+    def _ctx_with_function_call(call_id: str) -> ChatContext:
+        ctx = ChatContext.empty()
+        ctx.add_message(role="user", content="hello")
+        ctx.insert(FunctionCall(call_id=call_id, name="tool", arguments="{}"))
+        ctx.insert(FunctionCallOutput(call_id=call_id, name="tool", output="ok", is_error=False))
+        return ctx
+
+    @staticmethod
+    async def _drain(stream) -> None:
+        try:
+            async for _ in stream:
+                pass
+        finally:
+            await stream.aclose()
+
+    @pytest.mark.asyncio
+    async def test_gemini_25_sends_sentinel_when_cache_is_none(self) -> None:
+        llm = LLM(model="gemini-2.5-flash", api_key="test")
+        llm._thought_signatures = None
+
+        fake, captured = TestCachedContentRequestSuppression._patched_stream_capture()
+        with patch.object(llm._client.aio.models, "generate_content_stream", fake):
+            await self._drain(llm.chat(chat_ctx=self._ctx_with_function_call("call_from_openai")))
+
+        function_call_part = captured["contents"][1].parts[0]
+        assert function_call_part.thought_signature == b"skip_thought_signature_validator"
+
+    @pytest.mark.asyncio
+    async def test_pre_gemini_25_omits_thought_signature_even_with_cached_signature(self) -> None:
+        llm = LLM(model="gemini-2.0-flash", api_key="test")
+        llm._thought_signatures = {"call_1": b"real_signature"}
+
+        fake, captured = TestCachedContentRequestSuppression._patched_stream_capture()
+        with patch.object(llm._client.aio.models, "generate_content_stream", fake):
+            await self._drain(llm.chat(chat_ctx=self._ctx_with_function_call("call_1")))
+
+        function_call_part = captured["contents"][1].parts[0]
+        assert function_call_part.thought_signature is None
+
+
 class TestMediaResolution:
     def test_llm_media_resolution_is_passed_to_stream_kwargs(self):
         model = LLM(