Fix DeepSeek interleaved tool-call streaming

src/ava/llm/openai.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from ava.base import AvaError, CancelToken, ErrorKind
 from ava.base.cancel import NEVER
@@ -147,20 +147,24 @@ def openai_request_body(context: Context, model: str, effort: str | None) -> str
             messages.extend(_tool_messages(item))
     body["messages"] = messages
     if context.tools:
-        # The loop dispatches one streamed call at a time.
+        # Prefer single calls, but tolerate endpoints that ignore this hint.
         body["parallel_tool_calls"] = False
         body["tools"] = [_tool_schema(tool) for tool in context.tools]
     encoded = _dumps(body)
     check_request_limits(counter[0], len(encoded.encode("utf-8")))
     return encoded
 
 
+@dataclass(slots=True)
+class _ToolCall:
+    id: str = ""
+    name: str = ""
+    arguments: list[str] = field(default_factory=list)
+
+
 @dataclass(slots=True)
 class OpenAIStreamState:
-    tool_index: int | None = None
-    tool_id: str = ""
-    tool_name: str = ""
-    tool_started: bool = False
+    tools: dict[int, _ToolCall] = field(default_factory=dict)
     stop_reason: StopReason | None = None
 
 
@@ -196,51 +200,39 @@ def _emit_openai_usage(source: dict, sink: StreamSink) -> None:
         sink(StreamEvent(kind=StreamEventKind.usage, usage=usage))
 
 
-def _consume_tool_delta(delta: dict, sink: StreamSink, state: OpenAIStreamState) -> None:
-    index = delta.get("index")
-    if not isinstance(index, int):
+def _consume_tool_delta(delta: dict, state: OpenAIStreamState) -> None:
+    index = _int_or_none(delta.get("index"))
+    if index is None or index < 0:
         raise AvaError(
             ErrorKind.parse,
             "OpenAI tool call delta is missing its index; check endpoint compatibility",
         )
-    if state.tool_index is not None and state.tool_index != index:
-        raise AvaError(
-            ErrorKind.provider,
-            "OpenAI streamed parallel tool calls after Ava disabled them; check endpoint compatibility",
-        )
-    state.tool_index = index
+    call = state.tools.setdefault(index, _ToolCall())
     streamed_id = delta.get("id") or ""
     if streamed_id:
-        if state.tool_id and state.tool_id != streamed_id:
+        if call.id and call.id != streamed_id:
             raise AvaError(
                 ErrorKind.parse,
                 "OpenAI changed a streamed tool call id; check endpoint compatibility",
             )
-        state.tool_id = streamed_id
+        call.id = streamed_id
     raw_function = delta.get("function")
     function: dict = raw_function if isinstance(raw_function, dict) else {}
     name = function.get("name")
     if isinstance(name, str) and name:
-        if state.tool_name and state.tool_name != name:
+        if call.name and call.name != name:
             raise AvaError(
                 ErrorKind.parse, "OpenAI changed a streamed tool name; check endpoint compatibility"
             )
-        state.tool_name = name
-    if not state.tool_started and state.tool_id and state.tool_name:
-        sink(
-            StreamEvent(
-                kind=StreamEventKind.tool_call_start, id=state.tool_id, name=state.tool_name
-            )
-        )
-        state.tool_started = True
+        call.name = name
     arguments = function.get("arguments")
     if isinstance(arguments, str) and arguments:
-        if not state.tool_started:
+        if not call.id or not call.name:
             raise AvaError(
                 ErrorKind.parse,
                 "OpenAI streamed tool arguments before the call identity; check endpoint compatibility",
             )
-        sink(StreamEvent(kind=StreamEventKind.tool_call_delta, text=arguments, id=state.tool_id))
+        call.arguments.append(arguments)
 
 
 def _apply_finish_reason(reason: str, sink: StreamSink, state: OpenAIStreamState) -> None:
@@ -249,13 +241,27 @@ def _apply_finish_reason(reason: str, sink: StreamSink, state: OpenAIStreamState
     elif reason == "length":
         state.stop_reason = StopReason.max_tokens
     elif reason == "tool_calls":
-        if not state.tool_started:
+        calls = [state.tools[index] for index in sorted(state.tools)]
+        if not calls or any(not call.id or not call.name for call in calls):
             raise AvaError(
                 ErrorKind.parse,
                 "OpenAI stopped for tool calls without a complete call identity; check endpoint compatibility",
             )
-        sink(StreamEvent(kind=StreamEventKind.tool_call_end, id=state.tool_id))
-        state.tool_started = False
+        if len({call.id for call in calls}) != len(calls):
+            raise AvaError(ErrorKind.parse, "OpenAI returned duplicate tool call ids")
+        # The assembler accepts one open call at a time, even for interleaved wire deltas.
+        for call in calls:
+            sink(StreamEvent(kind=StreamEventKind.tool_call_start, id=call.id, name=call.name))
+            if call.arguments:
+                sink(
+                    StreamEvent(
+                        kind=StreamEventKind.tool_call_delta,
+                        text="".join(call.arguments),
+                        id=call.id,
+                    )
+                )
+            sink(StreamEvent(kind=StreamEventKind.tool_call_end, id=call.id))
+        state.tools.clear()
         state.stop_reason = StopReason.tool_use
     else:
         raise AvaError(ErrorKind.provider, f"OpenAI stopped with unsupported reason '{reason}'")
@@ -265,7 +271,7 @@ def consume_openai_event(
     event: SseEvent, sink: StreamSink, state: OpenAIStreamState
 ) -> StopReason | None:
     if event.data == "[DONE]":
-        if state.tool_started:
+        if state.tools:
             raise AvaError(
                 ErrorKind.parse,
                 "OpenAI stopped before finishing a tool call; retry or check endpoint compatibility",
@@ -310,7 +316,7 @@ def consume_openai_event(
             if isinstance(calls, list):
                 for call in calls:
                     if isinstance(call, dict):
-                        _consume_tool_delta(call, sink, state)
+                        _consume_tool_delta(call, state)
         finish_reason = choice.get("finish_reason")
         if isinstance(finish_reason, str) and finish_reason:
             _apply_finish_reason(finish_reason, sink, state)

tests/test_providers.py

@@ -184,6 +184,95 @@ async def test_openai_adapter_streams_and_normalizes_usage(fake_server: str):
     await provider.aclose()
 
 
+async def test_deepseek_interleaved_calls_execute_sequentially(
+    fake_server: str, home: Path, project: Path, monkeypatch: pytest.MonkeyPatch
+):
+    from ava.agent import Agent
+    from tests.conftest import message
+
+    def chunk(calls):
+        return (
+            "data: "
+            + json.dumps({"choices": [{"index": 0, "delta": {"tool_calls": calls}}]})
+            + "\n\n"
+        )
+
+    _Fake.stream = (
+        chunk(
+            [
+                {
+                    "index": 0,
+                    "id": "write_1",
+                    "function": {"name": "write", "arguments": '{"path":"parallel.txt",'},
+                },
+                {"index": 1, "id": "read_1", "function": {"name": "read", "arguments": '{"path":'}},
+            ]
+        )
+        + chunk([{"index": 1, "function": {"arguments": '"parallel.txt"}'}}])
+        + chunk([{"index": 0, "function": {"arguments": '"content":"sequential success"}'}}])
+        + 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n'
+        + "data: [DONE]\n\n"
+    )
+    original_post = _Fake.do_POST
+
+    def respond(handler):
+        if _Fake.requests:
+            _Fake.stream = (
+                'data: {"choices":[{"index":0,"delta":{"content":"Done"},"finish_reason":"stop"}]}\n\n'
+                "data: [DONE]\n\n"
+            )
+        original_post(handler)
+
+    monkeypatch.setattr(_Fake, "do_POST", respond)
+    provider = OpenAIProvider(Selection("deepseek", "deepseek-v4-flash"), fake_server, "key")
+    agent = Agent.create(provider, project)
+    try:
+        await agent.followup(message("Write parallel.txt and read it back."))
+        await agent.drive()
+        assert (project / "parallel.txt").read_text() == "sequential success"
+        assert len(_Fake.requests) == 2
+        assert _Fake.requests[0]["body"]["parallel_tool_calls"] is False
+        messages = _Fake.requests[1]["body"]["messages"]
+        calls = next(m["tool_calls"] for m in messages if m.get("tool_calls"))
+        assert [(c["id"], c["function"]["name"]) for c in calls] == [
+            ("write_1", "write"),
+            ("read_1", "read"),
+        ]
+        results = [m for m in messages if m["role"] == "tool"]
+        assert [m["tool_call_id"] for m in results] == ["write_1", "read_1"]
+        assert "sequential success" in results[1]["content"]
+    finally:
+        await agent.aclose()
+
+
+@pytest.mark.parametrize(
+    ("second_call", "finish", "error"),
+    [
+        ({"index": 1, "id": "c2", "function": {"name": "read"}}, False, "before finishing"),
+        ({"index": 1, "id": "c2"}, True, "without a complete call identity"),
+        ({"index": 1, "id": "c1", "function": {"name": "read"}}, True, "duplicate tool call ids"),
+        ({"index": 0, "id": "changed"}, True, "changed a streamed tool call id"),
+    ],
+)
+async def test_openai_rejects_invalid_buffered_calls(fake_server, second_call, finish, error):
+    calls = [
+        {"index": 0, "id": "c1", "function": {"name": "read", "arguments": "{}"}},
+        second_call,
+    ]
+    _Fake.stream = "data: " + json.dumps({"choices": [{"delta": {"tool_calls": calls}}]}) + "\n\n"
+    if finish:
+        _Fake.stream += 'data: {"choices":[{"finish_reason":"tool_calls"}]}\n\n'
+    _Fake.stream += "data: [DONE]\n\n"
+    provider = OpenAIProvider(Selection("deepseek", "deepseek-v4-flash"), fake_server, "key")
+    events = []
+    try:
+        with pytest.raises(AvaError, match=error):
+            await provider.stream(Context(), provider.selection, _collect(events))
+        assert events == []
+    finally:
+        await provider.aclose()
+
+
 def _tool() -> ToolDef:
     return ToolDef(
         "read",