fix(google): add fallback thought signature sentinel

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

@@ -63,9 +63,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 for every multi-turn
+            # function_call part. When a previous tool call came from another
+            # provider (for example through FallbackAdapter), we do not have a
+            # real signature, so use Google's documented validator-bypass sentinel.
+            if thought_signatures is not None:
+                fc_part["thought_signature"] = thought_signatures.get(
+                    msg.call_id, b"skip_thought_signature_validator"
+                )
             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

@@ -419,10 +419,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.
+            # New LLM instances may have no stored signatures yet; use an empty
+            # mapping so formatter can add the Gemini 2.5+ fallback sentinel.
+            thought_sigs = 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
             )
@@ -561,6 +563,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,12 +1,52 @@
 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,
     _requires_thought_signatures,
 )
 
 
+class TestGoogleThoughtSignatureFormatting:
+    def test_injects_existing_thought_signature_for_function_call(self):
+        ctx = ChatContext.empty()
+        ctx.add_message(role="user", content="hello")
+        ctx.insert(FunctionCall(call_id="call_1", name="tool", arguments="{}"))
+        ctx.insert(FunctionCallOutput(call_id="call_1", name="tool", output="ok", is_error=False))
+
+        turns, _ = ctx.to_provider_format(
+            format="google", thought_signatures={"call_1": b"real_signature"}
+        )
+
+        function_call_part = turns[1]["parts"][0]
+        assert function_call_part["thought_signature"] == b"real_signature"
+
+    def test_injects_skip_sentinel_for_missing_thought_signature(self):
+        ctx = ChatContext.empty()
+        ctx.add_message(role="user", content="hello")
+        ctx.insert(FunctionCall(call_id="call_from_openai", name="tool", arguments="{}"))
+        ctx.insert(
+            FunctionCallOutput(call_id="call_from_openai", name="tool", output="ok", is_error=False)
+        )
+
+        turns, _ = ctx.to_provider_format(format="google", thought_signatures={})
+
+        function_call_part = turns[1]["parts"][0]
+        assert function_call_part["thought_signature"] == b"skip_thought_signature_validator"
+
+    def test_omits_thought_signature_when_not_required(self):
+        ctx = ChatContext.empty()
+        ctx.add_message(role="user", content="hello")
+        ctx.insert(FunctionCall(call_id="call_1", name="tool", arguments="{}"))
+        ctx.insert(FunctionCallOutput(call_id="call_1", name="tool", output="ok", is_error=False))
+
+        turns, _ = ctx.to_provider_format(format="google")
+
+        function_call_part = turns[1]["parts"][0]
+        assert "thought_signature" not in function_call_part
+
+
 class TestGeminiModelDetection:
     """Tests for Gemini model detection helper functions."""
 

tests/test_plugin_google_llm.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -63,3 +64,16 @@ def test_empty_text_part_returns_none(self, llm_stream: LLMStream):
         chunk = llm_stream._parse_part("test-id", part)
 
         assert chunk is None
+
+    def test_stores_thought_signature_when_cache_is_none(self, llm_stream: LLMStream):
+        llm_stream._model = "gemini-2.5-flash"
+        llm_stream._llm._thought_signatures = None
+        part = SimpleNamespace(
+            function_call=SimpleNamespace(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"}