bug: Gemini streaming tool calls crash and lose thought signatures

Summary

Gemini's OpenAI-compatible Chat Completions stream does not always use the exact chunk shape that stream_openai_completions() currently assumes.

In particular, Gemini can:

• omit tool_calls[].index;
• split function arguments across indexless deltas;
• finish a tool-call response with finish_reason: "stop" rather than "tool_calls";
• return extra_content.google.thought_signature, including as a standalone metadata-only delta.

The current implementation indexes tc["index"] directly and therefore raises KeyError: 'index'. If that lookup alone is made optional, the tool call can still be lost at finish_reason: "stop". If both are fixed, subsequent tool rounds can still fail because extra_content is discarded when the call is normalized, persisted, and reconstructed for history.

This breaks Gemini tool use through an OpenAI-compatible connection. A tool can be selected correctly, but Computer aborts before executing it or receives a Gemini 4xx validation error on the follow-up request.

Reproduction

Reproduced against current main at 9c54711.

A minimal representative stream is:

data: {"choices":[{"index":0,"delta":{"tool_calls":[{"id":"function-call-1","type":"function","function":{"name":"get_weather","arguments":"{\"location\":\"Sydney\"}"},"extra_content":{"google":{"thought_signature":"REDACTED"}}}]},"finish_reason":null}]}
data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
data: [DONE]

At present, this reaches:

for tc in delta.get("tool_calls") or []:
    idx = tc["index"]  # KeyError: 'index'

Google's thought-signature rules also require the signature to be returned in the subsequent request exactly where it was received. For parallel function calls, it belongs to the first function call; all calls must precede all function responses. See Google's documentation: https://ai.google.dev/gemini-api/docs/generate-content/thought-signatures

Suggested fix

1. Merge tool-call deltas without assuming an index

Add a small accumulator in cptr/utils/ai.py. It preserves normal indexed OpenAI streams, matches indexless deltas by call ID, maps parallel indexless fragments by position, attaches a standalone Gemini signature to the first parallel call, and uses the last call only as the final single-call fallback:

def _merge_tool_call_delta(
    tool_calls: dict[int, dict],
    tool_call_delta: dict,
    last_index: int | None,
    *,
    position: int,
    batch_size: int,
) -> int:
    """Merge one OpenAI-compatible tool call delta and return its index."""
    index = tool_call_delta.get("index")
    call_id = tool_call_delta.get("id")

    if index is None and call_id:
        index = next(
            (
                existing_index
                for existing_index, current in tool_calls.items()
                if current.get("id") == call_id
            ),
            None,
        )
    if index is None and not call_id and batch_size > 1 and position in tool_calls:
        index = position

    extra_content = tool_call_delta.get("extra_content")
    google = extra_content.get("google") if isinstance(extra_content, dict) else None
    if (
        index is None
        and not tool_call_delta.get("function")
        and isinstance(google, dict)
        and google.get("thought_signature")
        and tool_calls
    ):
        index = next(iter(tool_calls))

    if index is None and last_index is not None:
        previous = tool_calls[last_index]
        if not call_id or not previous.get("id"):
            index = last_index
    if index is None:
        index = len(tool_calls)
        while index in tool_calls:
            index += 1

    current = tool_calls.setdefault(
        index,
        {"id": "", "name": "", "arguments_json": ""},
    )
    if call_id:
        current["id"] = call_id
    function = tool_call_delta.get("function") or {}
    if function.get("name"):
        current["name"] = function["name"]
    current["arguments_json"] += function.get("arguments", "")
    if "extra_content" in tool_call_delta:
        current["extra_content"] = copy.deepcopy(tool_call_delta["extra_content"])
    return index

Use it in the streaming loop and emit accumulated calls for either valid finish reason:

tool_calls: dict[int, dict] = {}
tool_calls_emitted = False
last_tool_call_index: int | None = None

# ... inside the chunk loop ...
tool_call_deltas = delta.get("tool_calls") or []
for position, tc in enumerate(tool_call_deltas):
    last_tool_call_index = _merge_tool_call_delta(
        tool_calls,
        tc,
        last_tool_call_index,
        position=position,
        batch_size=len(tool_call_deltas),
    )

finish_reason = choices[0].get("finish_reason") if choices else None
if (
    finish_reason in {"tool_calls", "stop"}
    and tool_calls
    and not tool_calls_emitted
):
    item = complete_reasoning_item()
    if item is not None:
        emitted = True
        yield {"type": "output", "item": item}
    for tc in tool_calls.values():
        event = {
            "type": "tool_call",
            "call_id": tc["id"],
            "name": tc["name"],
            "arguments": json.loads(tc["arguments_json"] or "{}"),
        }
        if "extra_content" in tc:
            event["extra_content"] = copy.deepcopy(tc["extra_content"])
        emitted = True
        yield event
    tool_calls_emitted = True

2. Preserve provider metadata through persistence and replay

When converting persisted function calls back to OpenAI messages in _output_items_to_messages():

if item.get("fc_id"):
    tc["fc_id"] = item["fc_id"]
if "extra_content" in item:
    tc["extra_content"] = item["extra_content"]

All four paths that create function-call items—automatic execution, queued approval, rejected/invalid calls, and ask_user—must retain the metadata. A helper avoids one path silently dropping it:

def _function_call_item(
    tool_call: dict,
    *,
    status: str,
    name: str | None = None,
    arguments: dict | None = None,
    **fields,
) -> dict:
    item = {
        "type": "function_call",
        "id": str(uuid.uuid4()),
        "call_id": tool_call["call_id"],
        "fc_id": tool_call.get("id", ""),
        "name": tool_call["name"] if name is None else name,
        "arguments": tool_call["arguments"] if arguments is None else arguments,
        "status": status,
        **fields,
    }
    if "extra_content" in tool_call:
        item["extra_content"] = tool_call["extra_content"]
    return item

Then replace the repeated inline function-call dictionaries with _function_call_item(...) in each path.

Tests

I reproduced this against the current main branch and have a focused standard-library test suite covering:

1. an indexless Gemini tool call ending with finish_reason: "stop";
2. ordinary indexed, fragmented OpenAI tool calls;
3. a single indexless fragmented call;
4. indexed parallel calls;
5. indexless fragmented parallel calls;
6. a standalone thought-signature delta attaching to the first parallel call;
7. persistence and replay of extra_content across a tool result.

All seven tests pass with the changes above, along with Ruff formatting and lint checks. I can provide the complete patch or a branch/PR if external contributions are accepted for this repository.

Start in cptr/utils/ai.py at stream_openai_completions() and inspect _output_items_to_messages() plus the four function-call item creation paths. Run the focused standard-library tests described in the issue, then verify indexed and indexless Gemini streams, finish_reason "stop", parallel calls, and thought-signature persistence and replay all pass with Ruff checks.