fix(providers): review round 3 — in-band error events, terminal-marker tolerance, adapter-owned de-fusion

Correctness:

- Responses _iter_stream handles the SDK's in-band `error` SSE event
  (ResponseErrorEvent is YIELDED, not raised, and no response.failed
  need follow): the real API code/message now surfaces — code-gated for
  retryability like response.failed — instead of the stream exhausting
  finish-less and hiding the cause behind a retried
  IncompleteStreamError.
- Anthropic message_stop supplies a missing stop_reason: it is a genuine
  terminal marker, so a compat /v1/messages shim whose message_delta
  omits stop_reason completes (blocks intact) rather than failing a
  generation that arrived — tolerance the retired non-streaming default
  provided, restored without weakening the died-mid-response gate.
- A Responses terminal event without its response payload still emits
  the finish reason its type implies (lax compat servers), losing only
  usage/blocks rather than the whole result.

Dispositions held (documented, not re-coded): the complete-or-error
gate stays for finish-less Chat Completions streams — indistinguishable
in-band from a died generation, and silent partial-storage is the worse
failure; CHANGELOG now names the shape and each provider's accepted
terminal markers. supports_streaming deletion and the stream_options
wire delta were ruled earlier and keep their release-note remediations.

Cleanup: index-degenerate de-fusion MOVED from drain_stream into the
chat adapter's iterator (mirroring the Anthropic iterator's index
assignment) so the interactive loop is fixed too and the drain returns
to a plain mirror of the main-loop accumulator; a parametrized test
locks "IncompleteStreamError is retryable" across all six provider
lanes instead of trusting per-adapter memory; scripted_anthropic_client
joins scripted_chat_client (shared _ScriptedClient class, no function
attrs) and the two remaining hand-rolled anthropic closures convert.

CHANGELOG.md

@@ -29,17 +29,25 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   blocking read that can hit client read-timeouts — the same reason the
   Anthropic adapter already streamed internally — and judge timeouts now
   *abort* the underlying HTTP read instead of abandoning a worker thread
-  on a dead call. Caveats: these lanes now carry the same
-  `stream_options: {include_usage: true}` the chat loop always sent —
-  OpenAI-compatible servers old enough to *ignore* it stop producing
-  usage rows on these lanes, and servers strict enough to *reject*
-  unknown fields (pre-2024 llama.cpp/proxy builds) will 400 — such a
-  server already couldn't serve turnstone's chat loop, but a
+  on a dead call. These lanes are also complete-or-error now: a stream
+  that ends without any finish signal is treated as a generation that
+  died mid-response and retried, instead of storing the partial text as
+  a clean result (previously a half-generated compaction summary could
+  silently replace real history). Caveats: these lanes now carry the
+  same `stream_options: {include_usage: true}` the chat loop always
+  sent — OpenAI-compatible servers old enough to *ignore* it stop
+  producing usage rows on these lanes, and servers strict enough to
+  *reject* unknown fields (pre-2024 llama.cpp/proxy builds) will 400 —
+  such a server already couldn't serve turnstone's chat loop, but a
   judge/utility alias pointed at one worked on 1.7 and needs to move to
-  a current server. Legacy models that reject streaming requests
-  outright (o1-era) likewise need a model alias pointing at a current
-  model — the unread `supports_streaming` capability flag (and its
-  admin tile) is gone.
+  a current server. A compat server whose streams never carry a finish
+  reason at all (each provider's terminal marker — Chat Completions'
+  final `finish_reason`, Anthropic's `message_stop`, Responses'
+  terminal event — all count) fails these lanes loudly for the same
+  reason; every maintained inference server sends one. Legacy models
+  that reject streaming requests outright (o1-era) likewise need a
+  model alias pointing at a current model — the unread
+  `supports_streaming` capability flag (and its admin tile) is gone.
 
 - **One turn interface for every model call: `core/model_turn.py` (#827).**
   Judges (intent + output guard), perception, title generation, compaction,

tests/_session_helpers.py

@@ -170,25 +170,39 @@ def fake_chat_stream(
     return chunks
 
 
-def scripted_chat_client(*scripts: Any) -> Any:
-    """A fake ``client.chat.completions.create`` that follows a script.
+class _ScriptedClient:
+    """Callable client-method fake following a script of stream builders.
 
     Call N returns the stream described by ``scripts[N]``; the last script
-    repeats for any further calls.  Each script is a dict of
-    :func:`fake_chat_stream` kwargs or a pre-built chunk list.  The
-    returned callable records every call's kwargs on ``.calls`` — read
-    ``len(fn.calls)`` where a test previously kept its own counter cell,
-    and ``fn.calls[i]["messages"]`` where it captured request bodies.
+    repeats for any further calls.  Each script is a dict of kwargs for
+    the bound stream builder, or a pre-built return value.  Records every
+    call's kwargs on ``.calls`` — read ``len(fn.calls)`` where a test
+    previously kept its own counter cell, and ``fn.calls[i]["messages"]``
+    where it captured request bodies.
     """
 
-    def _create(**kwargs: Any) -> Any:
-        _create.calls.append(kwargs)  # type: ignore[attr-defined]
-        i = min(len(_create.calls) - 1, len(scripts) - 1)  # type: ignore[attr-defined]
-        script = scripts[i]
-        return fake_chat_stream(**script) if isinstance(script, dict) else script
+    def __init__(self, scripts: tuple[Any, ...], to_stream: Any) -> None:
+        self._scripts = scripts
+        self._to_stream = to_stream
+        self.calls: list[dict[str, Any]] = []
 
-    _create.calls = []  # type: ignore[attr-defined]
-    return _create
+    def __call__(self, **kwargs: Any) -> Any:
+        self.calls.append(kwargs)
+        script = self._scripts[min(len(self.calls) - 1, len(self._scripts) - 1)]
+        return self._to_stream(**script) if isinstance(script, dict) else script
+
+
+def scripted_chat_client(*scripts: Any) -> _ScriptedClient:
+    """A scripted ``client.chat.completions.create`` — dict scripts are
+    :func:`fake_chat_stream` kwargs."""
+    return _ScriptedClient(scripts, fake_chat_stream)
+
+
+def scripted_anthropic_client(*scripts: Any) -> _ScriptedClient:
+    """A scripted ``client.messages.stream`` — dict scripts are
+    :func:`fake_anthropic_stream` kwargs (``blocks`` plus optional
+    ``stop_reason``/``usage``)."""
+    return _ScriptedClient(scripts, lambda **kw: fake_anthropic_stream(**kw))
 
 
 class FakeAnthropicBlock:

tests/test_drain_stream.py

@@ -123,34 +123,11 @@ class TestToolCallAssembly:
         )
         assert result.tool_calls[0]["id"] == ""
 
-    def test_index_degenerate_parallel_calls_get_distinct_slots(self):
-        # Historical compat servers (older vLLM, some llama.cpp builds)
-        # stream every parallel call at index 0 as whole deltas.  A delta
-        # whose id differs from its slot's opens a NEW call — without this,
-        # distinct calls fuse into one entry with concatenated garbage
-        # arguments.  Id-less fragments keep following their index's
-        # current call.
-        result = drain_stream(
-            iter(
-                [
-                    StreamChunk(
-                        tool_call_deltas=[
-                            ToolCallDelta(index=0, id="a", name="read", arguments_delta='{"p": 1}')
-                        ]
-                    ),
-                    StreamChunk(
-                        tool_call_deltas=[
-                            ToolCallDelta(index=0, id="b", name="write", arguments_delta='{"p": ')
-                        ]
-                    ),
-                    StreamChunk(tool_call_deltas=[ToolCallDelta(index=0, arguments_delta="2}")]),
-                    StreamChunk(finish_reason="tool_calls"),
-                ]
-            )
-        )
-        assert [tc["id"] for tc in result.tool_calls] == ["a", "b"]
-        assert result.tool_calls[0]["function"]["arguments"] == '{"p": 1}'
-        assert result.tool_calls[1]["function"]["arguments"] == '{"p": 2}'
+    # Index-degenerate parallel-call de-fusion lives in the CHAT ADAPTER's
+    # iterator (so the interactive loop is fixed too) — pinned in
+    # test_providers.py::TestOpenAIProvider::
+    # test_streaming_remaps_index_degenerate_parallel_calls.  The drain
+    # accumulates by index verbatim; adapters own index sanity.
 
 
 class TestUsageMerge:

tests/test_providers.py

@@ -743,6 +743,60 @@ class TestOpenAIProvider:
         assert result.tool_calls[0]["function"]["arguments"] == '{"path": "foo.py"}'
         assert result.finish_reason == "tool_calls"
 
+    def test_streaming_remaps_index_degenerate_parallel_calls(self) -> None:
+        # Historical compat servers (older vLLM, some llama.cpp builds)
+        # stream every parallel call at index 0 as whole deltas.  The
+        # iterator opens a new slot when a delta's id contradicts its
+        # index's current call, so BOTH consumers (drain_stream and the
+        # chat loop's accumulator) see distinct calls; id-less argument
+        # fragments keep following their index's current slot.
+        def _tc_chunk(tc_id: str, name: str, args: str) -> SimpleNamespace:
+            tc = SimpleNamespace(
+                index=0, id=tc_id, function=SimpleNamespace(name=name, arguments=args)
+            )
+            delta = SimpleNamespace(
+                content=None,
+                tool_calls=[tc],
+                reasoning=None,
+                reasoning_content=None,
+                annotations=None,
+            )
+            return SimpleNamespace(
+                choices=[SimpleNamespace(finish_reason=None, delta=delta)], usage=None
+            )
+
+        finish = SimpleNamespace(
+            choices=[
+                SimpleNamespace(
+                    finish_reason="tool_calls",
+                    delta=SimpleNamespace(
+                        content=None,
+                        tool_calls=None,
+                        reasoning=None,
+                        reasoning_content=None,
+                        annotations=None,
+                    ),
+                )
+            ],
+            usage=None,
+        )
+        chunks = [
+            _tc_chunk("a", "read", '{"p": 1}'),
+            _tc_chunk("b", "write", '{"p": 2}'),
+            finish,
+        ]
+        client = MagicMock()
+        client.chat.completions.create.return_value = chunks
+
+        result = drain_stream(
+            self.provider.create_streaming(
+                client=client, model="m", messages=[{"role": "user", "content": "x"}]
+            )
+        )
+        assert [tc["id"] for tc in result.tool_calls] == ["a", "b"]
+        assert result.tool_calls[0]["function"]["arguments"] == '{"p": 1}'
+        assert result.tool_calls[1]["function"]["arguments"] == '{"p": 2}'
+
     def test_drained_stream_usage(self) -> None:
         client = MagicMock()
         client.chat.completions.create.return_value = fake_chat_stream(
@@ -1174,6 +1228,36 @@ class TestAnthropicProvider:
         citations = result.provider_blocks[0].get("citations")
         assert citations == [{"type": "web_search_result_location", "url": "https://x.test"}]
 
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_message_stop_supplies_missing_stop_reason(self, mock_ensure: MagicMock) -> None:
+        # Compat tolerance: a /v1/messages shim that streams content and
+        # message_stop but never a message_delta stop_reason.  message_stop
+        # is a genuine terminal marker, so the drained stream completes
+        # (blocks intact) instead of failing a generation that arrived.
+        events = [
+            _anthropic_event("content_block_start", block_type="text", index=0),
+            _anthropic_event(
+                "content_block_delta", delta_type="text_delta", text="intact", index=0
+            ),
+            _anthropic_event("content_block_stop", index=0),
+            _anthropic_event("message_stop"),
+        ]
+        stream_ctx = MagicMock()
+        stream_ctx.__enter__ = MagicMock(return_value=iter(events))
+        stream_ctx.__exit__ = MagicMock(return_value=False)
+        client = MagicMock()
+        client.messages.stream.return_value = stream_ctx
+
+        result = drain_stream(
+            self.provider.create_streaming(
+                client=client,
+                model="claude-sonnet-4-20250514",
+                messages=[{"role": "user", "content": "hi"}],
+            )
+        )
+        assert result.content == "intact"
+        assert result.finish_reason == "stop"
+
     @patch("turnstone.core.providers._anthropic._ensure_anthropic")
     def test_drained_stream_usage(self, mock_ensure: MagicMock) -> None:
         client = MagicMock()
@@ -5265,6 +5349,36 @@ class TestResponsesDrainedStream:
         assert result.finish_reason == "length"
         assert [b["type"] for b in result.provider_blocks] == ["reasoning", "message"]
 
+    def test_error_event_surfaces_real_api_message(self) -> None:
+        # The SDK YIELDS in-band `error` SSE events (ResponseErrorEvent)
+        # rather than raising; without a branch the stream exhausts
+        # finish-less and the real API message hides behind a misleading
+        # IncompleteStreamError.  Deterministic codes stop retries.
+        events = [SimpleNamespace(type="error", code="invalid_request", message="bad tool schema")]
+        with pytest.raises(RuntimeError, match="bad tool schema"):
+            self._drain(events)
+
+    def test_transient_error_event_is_retryable(self) -> None:
+        from turnstone.core.providers._openai_responses import ResponsesStreamFailedError
+
+        events = [SimpleNamespace(type="error", code="server_error", message="overloaded")]
+        with pytest.raises(ResponsesStreamFailedError, match="overloaded"):
+            self._drain(events)
+
+    def test_terminal_event_without_payload_still_finishes(self) -> None:
+        # A lax compat server may emit the terminal event with no response
+        # payload — it is still a terminal signal, so the drained stream
+        # completes (without usage/blocks) instead of raising
+        # IncompleteStreamError over a generation that fully arrived.
+        events = [
+            SimpleNamespace(type="response.output_text.delta", delta="all here"),
+            SimpleNamespace(type="response.completed", response=None),
+        ]
+        result = self._drain(events)
+        assert result.content == "all here"
+        assert result.finish_reason == "stop"
+        assert result.usage is None
+
     def test_transient_failed_event_raises_typed_retryable_error(self) -> None:
         # A TRANSIENT in-band response.failed (server_error / rate limit)
         # raises the typed error the provider advertises as retryable —
@@ -5304,3 +5418,20 @@ class TestResponsesDrainedStream:
             self._drain(events)
         assert not isinstance(excinfo.value, ResponsesStreamFailedError)
         assert type(excinfo.value).__name__ not in self.provider.retryable_error_names
+
+
+class TestTransportRetryability:
+    """Every provider must advertise the shared transport error as
+    retryable — the drain raises IncompleteStreamError for ALL lanes, so a
+    provider omitting it silently loses retry-on-dead-stream (the
+    complete-or-error contract's second half)."""
+
+    @pytest.mark.parametrize(
+        "provider_name",
+        ["openai", "openai-compatible", "anthropic", "anthropic-compatible", "google", "xai"],
+    )
+    def test_incomplete_stream_error_is_retryable(self, provider_name: str) -> None:
+        from turnstone.core.providers import create_provider
+
+        provider = create_provider(provider_name)
+        assert "IncompleteStreamError" in provider.retryable_error_names

tests/test_session.py

@@ -14,8 +14,8 @@ import pytest
 from tests._session_helpers import (
     FakeAnthropicBlock,
     as_stream,
-    fake_anthropic_stream,
     mock_completion_result,
+    scripted_anthropic_client,
     scripted_chat_client,
 )
 from turnstone.core.session import _IMAGE_EXTENSIONS, _IMAGE_SIZE_CAP, ChatSession
@@ -2588,28 +2588,22 @@ class TestAgentChildRegistration:
         session._provider = AnthropicProvider()
         session.ui.note_agent_child = MagicMock()
 
-        seen: list[dict] = []
-        call_count = [0]
-
-        def fake_stream(**kwargs):
-            seen.append(kwargs)
-            call_count[0] += 1
-            if call_count[0] == 1:
-                return fake_anthropic_stream(
-                    [
-                        FakeAnthropicBlock(
-                            type="thinking", thinking="check the file first", signature="sig_v1"
-                        ),
-                        FakeAnthropicBlock(type="text", text="reading"),
-                        FakeAnthropicBlock(
-                            type="tool_use", id="toolu_01AB", name="read_file", input={"path": "x"}
-                        ),
-                    ],
-                    stop_reason="tool_use",
-                )
-            return fake_anthropic_stream([FakeAnthropicBlock(type="text", text="done")])
-
-        session.client.messages.stream = fake_stream
+        client_fn = scripted_anthropic_client(
+            {
+                "blocks": [
+                    FakeAnthropicBlock(
+                        type="thinking", thinking="check the file first", signature="sig_v1"
+                    ),
+                    FakeAnthropicBlock(type="text", text="reading"),
+                    FakeAnthropicBlock(
+                        type="tool_use", id="toolu_01AB", name="read_file", input={"path": "x"}
+                    ),
+                ],
+                "stop_reason": "tool_use",
+            },
+            {"blocks": [FakeAnthropicBlock(type="text", text="done")]},
+        )
+        session.client.messages.stream = client_fn
 
         def fake_prepare(tc_dict, **_kwargs):
             return {
@@ -2639,7 +2633,7 @@ class TestAgentChildRegistration:
         # Internal key stays minted — the nesting registry saw the "::" id.
         assert session.ui.note_agent_child.call_args.args[0] == "task-1::r1s1::toolu_01AB"
         # Second request: the assistant wire turn IS the native lane.
-        replay = seen[1]["messages"]
+        replay = client_fn.calls[1]["messages"]
         assistant = next(
             m for m in replay if m["role"] == "assistant" and isinstance(m.get("content"), list)
         )
@@ -2672,26 +2666,20 @@ class TestAgentChildRegistration:
         session._provider = AnthropicProvider()
         session.ui.note_agent_child = MagicMock()
 
-        seen: list[dict] = []
-        call_count = [0]
-
-        def fake_stream(**kwargs):
-            seen.append(kwargs)
-            call_count[0] += 1
-            if call_count[0] == 1:
-                return fake_anthropic_stream(
-                    [
-                        FakeAnthropicBlock(type="thinking", thinking="hm", signature="sig_b"),
-                        # Blank provider id — the back-fill case.
-                        FakeAnthropicBlock(
-                            type="tool_use", id="", name="read_file", input={"path": "x"}
-                        ),
-                    ],
-                    stop_reason="tool_use",
-                )
-            return fake_anthropic_stream([FakeAnthropicBlock(type="text", text="done")])
-
-        session.client.messages.stream = fake_stream
+        client_fn = scripted_anthropic_client(
+            {
+                "blocks": [
+                    FakeAnthropicBlock(type="thinking", thinking="hm", signature="sig_b"),
+                    # Blank provider id — the back-fill case.
+                    FakeAnthropicBlock(
+                        type="tool_use", id="", name="read_file", input={"path": "x"}
+                    ),
+                ],
+                "stop_reason": "tool_use",
+            },
+            {"blocks": [FakeAnthropicBlock(type="text", text="done")]},
+        )
+        session.client.messages.stream = client_fn
 
         def fake_prepare(tc_dict, **_kwargs):
             return {
@@ -2731,7 +2719,7 @@ class TestAgentChildRegistration:
         # The replay request carries the native lane verbatim — thinking and
         # signature intact — with tool_use and tool_result agreeing on the
         # manufactured id: no blank id, no orphan, no lost reasoning.
-        replay = seen[1]["messages"]
+        replay = client_fn.calls[1]["messages"]
         assistant = next(
             m for m in replay if m["role"] == "assistant" and isinstance(m.get("content"), list)
         )

turnstone/core/providers/_anthropic.py

@@ -926,6 +926,7 @@ class AnthropicProvider:
         # Capture raw content blocks for multi-turn preservation
         raw_blocks: dict[int, dict[str, Any]] = {}
         saw_text_block = False
+        emitted_finish = False
 
         for event in stream:
             sc = StreamChunk()
@@ -1064,10 +1065,23 @@ class AnthropicProvider:
                     )
                 if hasattr(event.delta, "stop_reason") and event.delta.stop_reason:
                     sc.finish_reason = _normalize_finish_reason(event.delta.stop_reason)
+                    emitted_finish = True
                     # Emit all raw content blocks for multi-turn preservation
                     if raw_blocks:
                         sc.provider_blocks = [raw_blocks[i] for i in sorted(raw_blocks)]
 
+            elif event_type == "message_stop":
+                # Terminal marker: the message completed even if the compat
+                # server's message_delta never carried a stop_reason (the
+                # official API always sends one; the retired non-streaming
+                # path defaulted missing stop_reason to end_turn).  Without
+                # this, the drain's complete-or-error gate fails a stream
+                # whose content actually arrived intact.
+                if not emitted_finish:
+                    sc.finish_reason = "stop"
+                    if raw_blocks:
+                        sc.provider_blocks = [raw_blocks[i] for i in sorted(raw_blocks)]
+
             elif event_type == "message_start":
                 if hasattr(event.message, "usage") and event.message.usage:
                     u = event.message.usage

turnstone/core/providers/_openai_chat.py

@@ -202,6 +202,16 @@ class OpenAIChatCompletionsProvider:
         tool_call_count = 0
         last_finish_reason: str | None = None
         completion_tokens: int | None = None
+        # Remap wire indexes onto logical slots: index-degenerate compat
+        # servers (historical vLLM/llama.cpp builds) emit every parallel
+        # tool call at index 0 — a delta whose id contradicts its index's
+        # current call opens a new slot, mirroring the Anthropic iterator's
+        # per-block index assignment.  Id-less argument fragments keep
+        # following their index's current slot, so downstream accumulators
+        # (drain_stream, the chat loop) can key by index safely.
+        slot_for_index: dict[int, int] = {}
+        slot_ids: dict[int, str] = {}
+        next_slot = 0
         for chunk in stream:
             sc = StreamChunk()
 
@@ -236,9 +246,20 @@ class OpenAIChatCompletionsProvider:
             # Tool calls
             if delta.tool_calls:
                 for tc_delta in delta.tool_calls:
-                    tcd = ToolCallDelta(index=tc_delta.index)
-                    if tc_delta.id:
-                        tcd.id = tc_delta.id
+                    wire_index = tc_delta.index
+                    tc_id = tc_delta.id or ""
+                    slot = slot_for_index.get(wire_index)
+                    if slot is None or (
+                        tc_id and slot_ids.get(slot, "") and slot_ids[slot] != tc_id
+                    ):
+                        slot = next_slot
+                        next_slot += 1
+                        slot_for_index[wire_index] = slot
+                    if tc_id:
+                        slot_ids[slot] = tc_id
+                    tcd = ToolCallDelta(index=slot)
+                    if tc_id:
+                        tcd.id = tc_id
                     if tc_delta.function:
                         if tc_delta.function.name:
                             tcd.name = tc_delta.function.name

turnstone/core/providers/_openai_responses.py

@@ -668,6 +668,15 @@ class OpenAIResponsesProvider:
             # finish reason, final usage, AND collected provider_blocks.
             if event_type in ("response.completed", "response.incomplete"):
                 response = getattr(event, "response", None)
+                if not response:
+                    # A terminal event without its payload (lax compat
+                    # server) is still a terminal signal — emit the finish
+                    # reason implied by the event type so the drain's
+                    # complete-or-error gate sees a completed stream, just
+                    # without usage/blocks.
+                    last_finish = "stop" if event_type == "response.completed" else "length"
+                    yield StreamChunk(finish_reason=last_finish)
+                    continue
                 if response:
                     status = getattr(response, "status", "completed")
                     last_finish = "stop" if status == "completed" else "length"
@@ -696,6 +705,20 @@ class OpenAIResponsesProvider:
                     yield sc
                 continue
 
+            # -- in-band error event (ResponseErrorEvent) --
+            # The SDK YIELDS `error` SSE events rather than raising, and no
+            # response.failed necessarily follows — without this branch the
+            # stream exhausts finish-less and the real API message is lost
+            # behind a misleading IncompleteStreamError.
+            if event_type == "error":
+                error_msg = getattr(event, "message", "Unknown error") or "Unknown error"
+                error_code = getattr(event, "code", "") or ""
+                if error_code in _TRANSIENT_FAILURE_CODES:
+                    raise ResponsesStreamFailedError(
+                        f"Responses API error ({error_code}): {error_msg}"
+                    )
+                raise RuntimeError(f"Responses API error ({error_code or 'unknown'}): {error_msg}")
+
             # -- error --
             if event_type == "response.failed":
                 response = getattr(event, "response", None)

turnstone/core/providers/_protocol.py

@@ -127,12 +127,9 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult:
       and completion tokens across separate events.
     - Tool calls accumulate by ``ToolCallDelta.index``: ``id``/``name``
       are whole values (last truthy wins), ``arguments_delta``
-      concatenates.  A delta carrying an id DIFFERENT from its slot's
-      opens a new call instead — index-degenerate compat servers
-      (historical vLLM/llama.cpp builds emit every parallel call at
-      index 0) would otherwise fuse distinct calls into garbage
-      arguments.  Deltas without ids keep routing to their index's
-      current call: fragments follow their call's announcement.
+      concatenates.  Adapters own index sanity — the chat iterator
+      remaps index-degenerate wire deltas onto distinct slots before
+      they reach any accumulator (this one or the chat loop's).
     - ``provider_blocks`` replaces on each non-empty emission — every
       adapter attaches its full block list exactly once, on or after the
       terminal chunk.
@@ -150,12 +147,7 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult:
     content_parts: list[str] = []
     reasoning_parts: list[str] = []
     trailing_info_parts: list[str] = []
-    # Slots in arrival order, each remembering its wire index — the result
-    # sorts by (index, arrival) so well-formed streams keep the array order
-    # the retired non-streaming body had, and collision-opened slots stay
-    # in arrival order behind their shared index.
-    tool_slots: list[tuple[int, dict[str, Any]]] = []
-    slot_for_index: dict[int, int] = {}
+    tool_calls_acc: dict[int, dict[str, Any]] = {}
     usage: UsageInfo | None = None
     finish_reason: str | None = None
     provider_blocks: list[dict[str, Any]] = []
@@ -166,19 +158,10 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult:
         if sc.reasoning_delta:
             reasoning_parts.append(sc.reasoning_delta)
         for tcd in sc.tool_call_deltas:
-            slot = slot_for_index.get(tcd.index)
-            if slot is None or (
-                tcd.id and tool_slots[slot][1]["id"] and tool_slots[slot][1]["id"] != tcd.id
-            ):
-                slot = len(tool_slots)
-                slot_for_index[tcd.index] = slot
-                tool_slots.append(
-                    (
-                        tcd.index,
-                        {"id": "", "type": "function", "function": {"name": "", "arguments": ""}},
-                    )
-                )
-            tc = tool_slots[slot][1]
+            tc = tool_calls_acc.setdefault(
+                tcd.index,
+                {"id": "", "type": "function", "function": {"name": "", "arguments": ""}},
+            )
             if tcd.id:
                 tc["id"] = tcd.id
             if tcd.name:
@@ -205,7 +188,7 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult:
     for info in trailing_info_parts:
         content += "\n\n" + info
 
-    tool_calls = [tc for _, tc in sorted(tool_slots, key=lambda pair: pair[0])]
+    tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)]
     return CompletionResult(
         content=content,
         tool_calls=tool_calls or None,