Skip to content

fix: bridge unsigned Gemini 3 tool histories #9853

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SunmiJJW wants to merge 2 commits into
base: master
Choose a base branch
from
+183 −12
Open

fix: bridge unsigned Gemini 3 tool histories #9853

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d57934
fix: bridge unsigned Gemini 3 tool histories
SunmiJJW Aug 27, 2026
821f2c7
fix: drop foreign Gemini tool-step signatures
SunmiJJW Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 36 additions & 12 deletions astrbot/core/provider/sources/gemini_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,10 @@ def append_or_extend(
else:
contents.append(content_cls(parts=part))

model_name = str(
payloads.get("model", getattr(self, "model_name", "")),
).lower()
is_gemini3_model = model_name.rsplit("/", 1)[-1].startswith("gemini-3")
gemini_contents: list[types.Content] = []
for message in payloads["messages"]:
role, content = message["role"], message.get("content")
Expand All @@ -351,6 +355,14 @@ def append_or_extend(

elif role == "assistant":
parts = []
tool_calls = message.get("tool_calls") or []
tool_calls_have_thought_signature = any(
isinstance(tool, dict)
and isinstance(tool.get("extra_content"), dict)
and isinstance(tool["extra_content"].get("google"), dict)
and tool["extra_content"]["google"].get("thought_signature")
for tool in tool_calls
)
if isinstance(content, str):
parts.append(types.Part.from_text(text=content))
elif isinstance(content, list):
Expand All @@ -373,17 +385,19 @@ def append_or_extend(
)
thinking_signature = None

if (
is_gemini3_model
and tool_calls
and not tool_calls_have_thought_signature
):
# An unsigned tool step came from another provider, so
# its opaque reasoning signature is not valid for Gemini.
thinking_signature = None

if (
not text
and thinking_signature
and "tool_calls" in message
and any(
isinstance(tool, dict)
and isinstance(tool.get("extra_content"), dict)
and isinstance(tool["extra_content"].get("google"), dict)
and tool["extra_content"]["google"].get("thought_signature")
for tool in message["tool_calls"]
)
and tool_calls_have_thought_signature
):
# If the main content is empty but tool calls have thought signatures,
# skip adding an empty text part to deduplicate the thinking signature in the main content and tool calls.
Expand All @@ -396,23 +410,33 @@ def append_or_extend(
)
)

if "tool_calls" in message:
for tool in message["tool_calls"]:
if tool_calls:
for index, tool in enumerate(tool_calls):
part = types.Part.from_function_call(
name=tool["function"]["name"],
args=json.loads(tool["function"]["arguments"]),
)
# we should set thought_signature back to part if exists
# for more info about thought_signature, see:
# https://ai.google.dev/gemini-api/docs/thought-signatures
ts_bs64 = None
if "extra_content" in tool and tool["extra_content"]:
ts_bs64 = (
tool["extra_content"]
.get("google", {})
.get("thought_signature")
)
if ts_bs64:
part.thought_signature = base64.b64decode(ts_bs64)
if ts_bs64:
part.thought_signature = base64.b64decode(ts_bs64)
elif (
is_gemini3_model
and not tool_calls_have_thought_signature
and index == 0
Comment thread
SunmiJJW marked this conversation as resolved.
):
# Cross-provider histories do not carry Gemini's opaque
# signature. Gemini 3 accepts this documented sentinel
# on the first function call in the step.
part.thought_signature = b"skip_thought_signature_validator"
parts.append(part)

if not parts:
Expand Down
147 changes: 147 additions & 0 deletions tests/test_gemini_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import base64
from types import SimpleNamespace

import httpx
Expand Down Expand Up @@ -107,6 +108,152 @@ async def test_gemini_prepare_conversation_resolves_local_history_image(tmp_path
assert image_part.inline_data.data == image_bytes


@pytest.mark.asyncio
async def test_gemini3_prepare_conversation_adds_signature_to_cross_provider_tool_calls():
provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI)

contents = await provider._prepare_conversation(
{
"model": "gemini-3.1-pro-preview",
"messages": [
{"role": "user", "content": "find the latest result"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "result"}',
},
"extra_content": {"openai": {"item_id": "item_1"}},
},
{
"function": {
"name": "read_page",
"arguments": '{"id": "page"}',
}
},
],
},
],
}
)

assert contents[-1].parts is not None
assert (
contents[-1].parts[0].thought_signature == b"skip_thought_signature_validator"
)
assert contents[-1].parts[1].thought_signature is None


@pytest.mark.asyncio
async def test_gemini3_prepare_conversation_drops_foreign_tool_step_signature():
provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI)
foreign_signature = base64.b64encode(b"foreign-provider-signature").decode("utf-8")

contents = await provider._prepare_conversation(
{
"model": "gemini-3.1-pro-preview",
"messages": [
{"role": "user", "content": "find the latest result"},
{
"role": "assistant",
"content": [
{
"type": "think",
"text": "provider-specific reasoning",
"encrypted": foreign_signature,
},
{"type": "text", "text": "I will check."},
],
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "result"}',
}
}
],
},
],
}
)

assert contents[-1].parts is not None
assert contents[-1].parts[0].text == "I will check."
assert contents[-1].parts[0].thought_signature is None
assert (
contents[-1].parts[1].thought_signature == b"skip_thought_signature_validator"
)


@pytest.mark.asyncio
async def test_gemini3_prepare_conversation_preserves_real_tool_signature():
provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI)
signature = b"real-gemini-signature"

contents = await provider._prepare_conversation(
{
"model": "gemini-3-flash-preview",
"messages": [
{"role": "user", "content": "find the latest result"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "result"}',
},
"extra_content": {
"google": {
"thought_signature": base64.b64encode(
signature
).decode("utf-8")
}
},
}
],
},
],
}
)

assert contents[-1].parts is not None
assert contents[-1].parts[0].thought_signature == signature


@pytest.mark.asyncio
async def test_gemini25_prepare_conversation_keeps_unsigned_tool_calls_unchanged():
provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI)

contents = await provider._prepare_conversation(
{
"model": "gemini-2.5-pro",
"messages": [
{"role": "user", "content": "find the latest result"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "result"}',
}
}
],
},
],
}
)

assert contents[-1].parts is not None
assert contents[-1].parts[0].thought_signature is None


def test_gemini_empty_output_raises_empty_model_output_error():
llm_response = LLMResponse(role="assistant")

Expand Down