fix(providers): review round 7 — id-disciplined slots, all-lane finish tolerance, retry backoff

Correctness:
- ToolCallSlotter: a slot whose id is KNOWN never splits on an id-less
  delta — on an id-disciplined server new calls arrive with ids, so an
  id-less fragment (the call's FIRST name announcement included) is
  always a continuation. Round-6 regression: {id} → {name} → {args}
  emission split into an unnamed id-bearing call plus a nameless twin.
  Also: a name arriving for a slot with no name yet never splits
  (args-first emission), and a bare same-name delta after complete
  arguments merges as a redundant footer instead of minting a phantom
  zero-argument call that would re-run a side-effecting tool.
- finish_reason_optional is honored on every drained lane, not just
  Chat Completions: Anthropic shims a missing message_delta
  stop_reason + message_stop pair, Responses a missing terminal event
  (both with collected blocks riding the shimmed finish) — the
  documented capabilities-JSON remediation now works on the
  anthropic-compatible/responses-compat gateways it was written for,
  matching the retired non-streaming paths' tolerance.
- Responses: an in-band error/response.failed frame arriving AFTER the
  terminal event is teardown noise — log and end the stream instead of
  raising away a generation already in hand (the in-band twin of
  drain_stream's post-finish transport-blip tolerance).
- model_turn drain retries pace like the SDK request retry they
  replace: 0.5s base, doubling, ±50% jitter — instant re-issues
  re-hit the still-active rate limit/overload and synchronize into
  fleet-scale retry bursts.
- Responses slot bookkeeping survives lax servers: slots minted by a
  counter (len(dict) collided calls after a duplicate/empty item-id
  overwrite), orphan argument deltas route to the most recently
  announced call instead of hardwired slot 0.

Cleanup:
- on_tool_call_delta now receives the normalized ToolCallDelta plus the
  raw SDK delta — Google's capture accumulates the exact bytes the
  mirror sees (the byte-identical extraction no longer exists twice).
- _ArgsScanner feeds only fully id-less slots (its verdict is never
  consulted for id'd slots — dominant-case hot path).
- Anthropic retryable set hoisted to a class constant (per-access
  frozenset allocation, same pattern already fixed on Responses).
- GoogleProvider class docstring names the hook-based capture instead
  of the deleted _extract_tool_calls override.

CHANGELOG.md

@@ -41,18 +41,19 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   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. Transient mid-stream deaths (connection drop, proxy
-  hiccup) are re-issued in place up to twice — the retry the SDK's
-  request loop used to provide these lanes invisibly. Each lane accepts
-  its own terminal marker (Anthropic `message_stop`, Responses terminal
-  events); on the Chat Completions lane a server that never sends
-  `finish_reason` at all needs `{"finish_reason_optional": true}` in the
-  model definition's capabilities JSON, which restores 1.7's tolerance
-  (clean end-of-stream after output = completion) for that model —
-  without it such streams fail as died-mid-generation, because SSE gives
-  no way to tell the two apart and the default favors catching
-  truncation. The unread `supports_streaming` capability flag (and its
-  admin tile) is gone; the o-series models it described are dropped from
-  the capability table entirely (see Removed).
+  hiccup) are re-issued in place up to twice with exponential backoff —
+  the retry the SDK's request loop used to provide these lanes
+  invisibly. Each lane accepts its own terminal marker (Anthropic
+  `message_stop`, Responses terminal events); a lax server/gateway that
+  never sends any terminal signal needs
+  `{"finish_reason_optional": true}` in the model definition's
+  capabilities JSON, which restores 1.7's tolerance (clean end-of-stream
+  after output = completion) for that model on every lane — without it
+  such streams fail as died-mid-generation, because SSE gives no way to
+  tell the two apart and the default favors catching truncation. The
+  unread `supports_streaming` capability flag (and its admin tile) is
+  gone; the o-series models it described are dropped from the capability
+  table entirely (see Removed).
 
 - **One turn interface for every model call: `core/model_turn.py` (#827).**
   Judges (intent + output guard), perception, title generation, compaction,

tests/_session_helpers.py

@@ -222,7 +222,7 @@ class FakeAnthropicBlock:
 def fake_anthropic_stream(
     blocks: list[Any],
     *,
-    stop_reason: str = "end_turn",
+    stop_reason: str | None = "end_turn",
     usage: Any = None,
 ) -> Any:
     """Fake Anthropic SDK stream context manager for tests that drive the
@@ -240,6 +240,10 @@ def fake_anthropic_stream(
     ``stop_reason`` (+ optional usage object).  Without the stripping, the
     provider's raw-block accumulator would double every text/thinking
     field (start capture + delta append).
+
+    ``stop_reason=None`` omits the closing ``message_delta`` entirely —
+    the terminal-signal-less lax-gateway shape ``finish_reason_optional``
+    exists for (content arrives, then the stream just ends).
     """
     events: list[Any] = []
     for idx, block in enumerate(blocks):
@@ -295,11 +299,12 @@ def fake_anthropic_stream(
                 )
             )
         events.append(SimpleNamespace(type="content_block_stop", index=idx))
-    events.append(
-        SimpleNamespace(
-            type="message_delta", usage=usage, delta=SimpleNamespace(stop_reason=stop_reason)
+    if stop_reason is not None or usage is not None:
+        events.append(
+            SimpleNamespace(
+                type="message_delta", usage=usage, delta=SimpleNamespace(stop_reason=stop_reason)
+            )
         )
-    )
 
     mgr = MagicMock()
     mgr.__enter__ = MagicMock(return_value=events)

tests/test_model_turn.py

@@ -103,10 +103,11 @@ class _FlakyProvider:
         return _iter()
 
 
-def test_model_turn_retries_transient_mid_stream_death() -> None:
+def test_model_turn_retries_transient_mid_stream_death(monkeypatch: pytest.MonkeyPatch) -> None:
     # The retired non-streaming transport read the whole body inside the
     # SDK's retried request, so single-shot lanes never saw a mid-body wire
     # blip — the drain-scoped loop is that retry's new home.
+    monkeypatch.setattr("turnstone.core.model_turn._DRAIN_RETRY_BASE_DELAY", 0.0)
     provider = _FlakyProvider(
         [
             IncompleteStreamError("stream died mid-response"),
@@ -121,7 +122,8 @@ def test_model_turn_retries_transient_mid_stream_death() -> None:
     assert len(provider.calls) == 2
 
 
-def test_model_turn_gives_up_after_retry_budget() -> None:
+def test_model_turn_gives_up_after_retry_budget(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("turnstone.core.model_turn._DRAIN_RETRY_BASE_DELAY", 0.0)
     provider = _FlakyProvider([IncompleteStreamError(f"death {i}") for i in range(5)])
     lane = ModelLane(provider=provider, client=object(), model="m")
 
@@ -132,6 +134,31 @@ def test_model_turn_gives_up_after_retry_budget() -> None:
     assert len(provider.calls) == 3
 
 
+def test_model_turn_retry_backs_off_between_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Instant re-issues are guaranteed to re-hit a still-active rate
+    # limit/overload — the loop paces like the SDK request retry it
+    # replaces: 0.5s base, doubling, ±50% jitter.
+    import turnstone.core.model_turn as model_turn_mod
+
+    sleeps: list[float] = []
+    monkeypatch.setattr(model_turn_mod, "time", SimpleNamespace(sleep=sleeps.append))
+    provider = _FlakyProvider(
+        [
+            IncompleteStreamError("death 1"),
+            IncompleteStreamError("death 2"),
+            CompletionResult(content="ok"),
+        ]
+    )
+    lane = ModelLane(provider=provider, client=object(), model="m")
+
+    result = model_turn(lane, [Turn.user("x")])
+
+    assert result.content == "ok"
+    assert len(sleeps) == 2
+    assert 0.25 <= sleeps[0] <= 0.75  # 0.5 * jitter[0.5, 1.5)
+    assert 0.5 <= sleeps[1] <= 1.5  # 1.0 * jitter[0.5, 1.5)
+
+
 def test_model_turn_does_not_retry_unrecognized_errors() -> None:
     provider = _FlakyProvider([RuntimeError("schema violation")])
     lane = ModelLane(provider=provider, client=object(), model="m")

tests/test_providers.py

@@ -967,6 +967,49 @@ class TestOpenAIProvider:
         assert [tc["function"]["name"] for tc in result.tool_calls] == ["refresh_state", "read"]
         assert result.tool_calls[1]["function"]["arguments"] == '{"p": "x"}'
 
+    def test_id_first_fragmented_call_stays_one_call(self) -> None:
+        # id → name → args across three fragments (the later two id-less):
+        # a slot with a KNOWN id never splits on id-less continuations —
+        # the call's first name fragment is not a re-announcement (round-7
+        # regression: it split into an unnamed id-bearing call plus a
+        # nameless-id twin).
+        result = self._drain_chunks(
+            [
+                _openai_stream_chunk(tool_calls=[_openai_tool_call_delta(index=0, tc_id="call_1")]),
+                _openai_stream_chunk(
+                    tool_calls=[_openai_tool_call_delta(index=0, name="get_weather")]
+                ),
+                _openai_stream_chunk(
+                    tool_calls=[_openai_tool_call_delta(index=0, arguments='{"city": "x"}')]
+                ),
+                _openai_stream_chunk(finish_reason="tool_calls"),
+            ]
+        )
+        assert len(result.tool_calls) == 1
+        assert result.tool_calls[0]["id"] == "call_1"
+        assert result.tool_calls[0]["function"]["name"] == "get_weather"
+        assert result.tool_calls[0]["function"]["arguments"] == '{"city": "x"}'
+
+    def test_idless_bare_name_footer_after_complete_args_merges(self) -> None:
+        # A bare same-name delta after the argument JSON closed is a
+        # redundant footer, not a second zero-argument call — splitting
+        # would run the side-effecting tool twice.
+        result = self._drain_chunks(
+            [
+                _openai_stream_chunk(
+                    tool_calls=[
+                        _openai_tool_call_delta(index=0, name="write_file", arguments='{"x": 1}')
+                    ]
+                ),
+                _openai_stream_chunk(
+                    tool_calls=[_openai_tool_call_delta(index=0, name="write_file")]
+                ),
+                _openai_stream_chunk(finish_reason="tool_calls"),
+            ]
+        )
+        assert len(result.tool_calls) == 1
+        assert result.tool_calls[0]["function"]["arguments"] == '{"x": 1}'
+
     def test_idless_name_first_then_args_with_repeated_name_merges(self) -> None:
         # Name announced first (no arguments), then arguments arrive
         # carrying the SAME name again: one call whose arguments are
@@ -1329,6 +1372,51 @@ class TestAnthropicProvider:
         assert result.tool_calls is None
         assert result.finish_reason == "stop"
 
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_terminal_signal_less_stream_raises_by_default(self, mock_ensure: MagicMock) -> None:
+        # No message_delta stop_reason and no message_stop: on a
+        # signal-disciplined server (the real API always sends both) this
+        # is a generation that died mid-response — the drain refuses to
+        # bless possibly-truncated content.
+        from turnstone.core.providers import IncompleteStreamError
+
+        client = MagicMock()
+        client.messages.stream.return_value = fake_anthropic_stream(
+            [SimpleNamespace(type="text", text="full answer")], stop_reason=None
+        )
+        with pytest.raises(IncompleteStreamError):
+            drain_stream(
+                self.provider.create_streaming(
+                    client=client,
+                    model="claude-sonnet-4-20250514",
+                    messages=[{"role": "user", "content": "hi"}],
+                )
+            )
+
+    @patch("turnstone.core.providers._anthropic._ensure_anthropic")
+    def test_terminal_signal_less_stream_completes_with_declared_tolerance(
+        self, mock_ensure: MagicMock
+    ) -> None:
+        # ``finish_reason_optional`` (operator-declared: this gateway never
+        # sends terminal signals) restores the retired non-streaming
+        # path's absent-stop_reason tolerance — the raw blocks ride the
+        # shimmed finish chunk.
+        client = MagicMock()
+        client.messages.stream.return_value = fake_anthropic_stream(
+            [SimpleNamespace(type="text", text="full answer")], stop_reason=None
+        )
+        result = drain_stream(
+            self.provider.create_streaming(
+                client=client,
+                model="claude-sonnet-4-20250514",
+                messages=[{"role": "user", "content": "hi"}],
+                capabilities=ModelCapabilities(finish_reason_optional=True),
+            )
+        )
+        assert result.content == "full answer"
+        assert result.finish_reason == "stop"
+        assert result.provider_blocks
+
     @patch("turnstone.core.providers._anthropic._ensure_anthropic")
     def test_drained_stream_with_tool_use(self, mock_ensure: MagicMock) -> None:
         client = MagicMock()
@@ -5472,12 +5560,15 @@ class TestResponsesDrainedStream:
         )
         return events
 
-    def _drain(self, events: list[Any]):
+    def _drain(self, events: list[Any], capabilities: ModelCapabilities | None = None):
         client = MagicMock()
         client.responses.create.return_value = events
         return drain_stream(
             self.provider.create_streaming(
-                client=client, model="gpt-5.1", messages=[{"role": "user", "content": "hi"}]
+                client=client,
+                model="gpt-5.1",
+                messages=[{"role": "user", "content": "hi"}],
+                capabilities=capabilities,
             )
         )
 
@@ -5512,6 +5603,115 @@ class TestResponsesDrainedStream:
         assert result.usage is not None
         assert result.provider_blocks
 
+    def test_terminal_event_less_stream_raises_by_default(self) -> None:
+        # No response.completed/incomplete ever arrived: on an
+        # event-disciplined server this is a generation that died
+        # mid-response — the drain refuses to bless possibly-truncated
+        # content.
+        from turnstone.core.providers import IncompleteStreamError
+
+        events = self._make_events(text="Hello")[:-1]  # drop the terminal event
+        with pytest.raises(IncompleteStreamError):
+            self._drain(events)
+
+    def test_terminal_event_less_stream_completes_with_declared_tolerance(self) -> None:
+        # ``finish_reason_optional`` (operator-declared: this server never
+        # sends terminal events) completes the clean output-bearing end —
+        # the .done-collected blocks ride the shimmed finish chunk.
+        events = self._make_events(text="Hello")[:-1]
+        result = self._drain(events, capabilities=ModelCapabilities(finish_reason_optional=True))
+        assert result.content == "Hello"
+        assert result.finish_reason == "stop"
+        assert result.provider_blocks
+
+    def test_post_terminal_error_event_keeps_completed_result(self) -> None:
+        # A trailing in-band error frame after response.completed is
+        # teardown noise — raising would discard a generation already in
+        # hand (the in-band twin of drain_stream's post-finish
+        # transport-blip tolerance).
+        events = self._make_events(text="Hello")
+        events.append(SimpleNamespace(type="error", code="server_error", message="boom"))
+        result = self._drain(events)
+        assert result.content == "Hello"
+        assert result.finish_reason == "stop"
+
+    def test_post_terminal_failed_event_keeps_completed_result(self) -> None:
+        events = self._make_events(text="Hello")
+        events.append(
+            SimpleNamespace(
+                type="response.failed",
+                response=SimpleNamespace(
+                    error=SimpleNamespace(code="server_error", message="boom")
+                ),
+            )
+        )
+        result = self._drain(events)
+        assert result.content == "Hello"
+        assert result.finish_reason == "stop"
+
+    def test_orphan_argument_deltas_route_to_last_announced_call(self) -> None:
+        # A lax server whose argument deltas reference an item_id that was
+        # never announced: they belong to the call most recently opened,
+        # not hardwired slot 0.
+        item_a = SimpleNamespace(type="function_call", call_id="call_a", id="item_a", name="alpha")
+        item_a.model_dump = lambda **_kw: {  # type: ignore[method-assign]
+            "type": "function_call",
+            "call_id": "call_a",
+            "name": "alpha",
+        }
+        item_b = SimpleNamespace(type="function_call", call_id="call_b", id="item_b", name="beta")
+        item_b.model_dump = lambda **_kw: {  # type: ignore[method-assign]
+            "type": "function_call",
+            "call_id": "call_b",
+            "name": "beta",
+        }
+        events = [
+            SimpleNamespace(type="response.output_item.added", item=item_a),
+            SimpleNamespace(type="response.output_item.added", item=item_b),
+            SimpleNamespace(
+                type="response.function_call_arguments.delta",
+                item_id="bogus",
+                delta='{"x": 1}',
+            ),
+            SimpleNamespace(
+                type="response.completed",
+                response=SimpleNamespace(status="completed", usage=None),
+            ),
+        ]
+        result = self._drain(events)
+        assert result.tool_calls is not None
+        by_name = {tc["function"]["name"]: tc["function"]["arguments"] for tc in result.tool_calls}
+        assert by_name["beta"] == '{"x": 1}'
+        assert by_name["alpha"] == ""
+
+    def test_duplicate_item_ids_keep_distinct_slots(self) -> None:
+        # Slot numbering must survive duplicate/empty item ids: len(dict)
+        # numbering collided the third call onto the second's slot once an
+        # overwrite kept the dict size flat.
+        def _item(call_id: str, item_id: str, name: str) -> SimpleNamespace:
+            item = SimpleNamespace(type="function_call", call_id=call_id, id=item_id, name=name)
+            item.model_dump = lambda **_kw: {  # type: ignore[method-assign]
+                "type": "function_call",
+                "call_id": call_id,
+                "name": name,
+            }
+            return item
+
+        events = [
+            SimpleNamespace(type="response.output_item.added", item=_item("call_a", "", "alpha")),
+            SimpleNamespace(type="response.output_item.added", item=_item("call_b", "", "beta")),
+            SimpleNamespace(
+                type="response.output_item.added", item=_item("call_c", "item_c", "gamma")
+            ),
+            SimpleNamespace(
+                type="response.completed",
+                response=SimpleNamespace(status="completed", usage=None),
+            ),
+        ]
+        result = self._drain(events)
+        assert result.tool_calls is not None
+        assert [tc["function"]["name"] for tc in result.tool_calls] == ["alpha", "beta", "gamma"]
+
     def test_completed_terminal_keeps_done_items_without_rebuild(self) -> None:
         # Happy path: every item got its ``output_item.done`` and the
         # terminal payload carries the same number of items — the rebuild

turnstone/core/model_turn.py

@@ -31,6 +31,8 @@ Contract, held deliberately narrow:
 
 from __future__ import annotations
 
+import random
+import time
 import uuid
 from dataclasses import dataclass, fields, replace
 from typing import TYPE_CHECKING, Any
@@ -65,6 +67,13 @@ log = get_logger(__name__)
 # retry (openai/anthropic default ``max_retries=2``) that covered the
 # whole body read on the retired non-streaming transport.
 _DRAIN_RETRIES = 2
+# Base for the exponential inter-attempt delay (0.5s → 1s, ±50% jitter) —
+# the SDK retry's pacing, minus Retry-After (an in-band stream failure
+# carries no header to honor).  An instant re-issue is guaranteed to
+# re-hit a still-active rate limit or overload, and at fleet scale
+# synchronized re-issues amplify the very condition being retried
+# through.  Module-level so tests can zero it.
+_DRAIN_RETRY_BASE_DELAY = 0.5
 
 # Block types that carry model reasoning natively.  Anthropic emits
 # ``thinking``/``redacted_thinking`` blocks, OpenAI Responses emits
@@ -750,12 +759,16 @@ def model_turn(
                 or type(exc).__name__ not in lane.provider.retryable_error_names
             ):
                 raise
+            delay = _DRAIN_RETRY_BASE_DELAY * (2 ** (attempt - 1)) * (0.5 + random.random())
             log.warning(
                 "model_turn.drain_retry",
                 error_type=type(exc).__name__,
                 attempt=attempt,
                 model=lane.model,
+                retry_in=round(delay, 2),
             )
+            if delay > 0:
+                time.sleep(delay)
 
     raw_calls: list[dict[str, Any]] = list(result.tool_calls or [])
     # Record blanks BEFORE the uuid back-fill: a back-filled id exists only

turnstone/core/providers/_anthropic.py

@@ -903,20 +903,35 @@ class AnthropicProvider:
             raise
         if cancel_ref is not None:
             cancel_ref.append(stream)
-        return self._iter_with_cleanup(stream, manager)
+        return self._iter_with_cleanup(
+            stream, manager, finish_reason_optional=caps.finish_reason_optional
+        )
 
-    def _iter_with_cleanup(self, stream: Any, manager: Any) -> Iterator[StreamChunk]:
+    def _iter_with_cleanup(
+        self, stream: Any, manager: Any, *, finish_reason_optional: bool = False
+    ) -> Iterator[StreamChunk]:
         """Iterate the Anthropic stream, ensuring the context manager exits."""
         try:
-            yield from self._iter_anthropic_stream(stream)
+            yield from self._iter_anthropic_stream(
+                stream, finish_reason_optional=finish_reason_optional
+            )
         except BaseException:
             manager.__exit__(*sys.exc_info())
             raise
         else:
             manager.__exit__(None, None, None)
 
-    def _iter_anthropic_stream(self, stream: Any) -> Iterator[StreamChunk]:
-        """Convert Anthropic streaming events to normalized StreamChunks."""
+    def _iter_anthropic_stream(
+        self, stream: Any, *, finish_reason_optional: bool = False
+    ) -> Iterator[StreamChunk]:
+        """Convert Anthropic streaming events to normalized StreamChunks.
+
+        *finish_reason_optional* is the model capability of the same name:
+        a lax anthropic-compatible gateway that ends the stream without
+        EITHER terminal signal (no ``message_delta`` stop_reason, no
+        ``message_stop``) gets the end-of-generator shim below — the
+        retired non-streaming path's absent-stop_reason tolerance.
+        """
         first = True
         # Map content block index → tool call index for our accumulator
         tool_block_to_index: dict[int, int] = {}
@@ -927,6 +942,7 @@ class AnthropicProvider:
         raw_blocks: dict[int, dict[str, Any]] = {}
         saw_text_block = False
         emitted_finish = False
+        delivered_output = False
 
         for event in stream:
             sc = StreamChunk()
@@ -1079,6 +1095,7 @@ class AnthropicProvider:
                 # whose content actually arrived intact.
                 if not emitted_finish:
                     sc.finish_reason = "stop"
+                    emitted_finish = True
                     if raw_blocks:
                         sc.provider_blocks = [raw_blocks[i] for i in sorted(raw_blocks)]
 
@@ -1098,29 +1115,51 @@ class AnthropicProvider:
                     )
 
             has_content = sc.content_delta or sc.reasoning_delta or sc.tool_call_deltas
-            if has_content and first:
-                sc.is_first = True
-                first = False
+            if has_content:
+                delivered_output = True
+                if first:
+                    sc.is_first = True
+                    first = False
 
             if has_content or sc.finish_reason or sc.usage or sc.info_delta:
                 yield sc
 
+        # Terminal-signal-less lax-gateway tolerance, armed ONLY by the
+        # operator-declared ``finish_reason_optional`` capability: a
+        # stream that ended cleanly after delivering output but carried
+        # NEITHER terminal signal (no message_delta stop_reason, no
+        # message_stop — an anthropic-compatible proxy shape; the real
+        # API always sends both) is a completed generation.  Everywhere
+        # else the drain's complete-or-error gate raises, because a
+        # missing message_stop on a signal-disciplined server means the
+        # generation died mid-response.
+        if finish_reason_optional and not emitted_finish and delivered_output:
+            sc = StreamChunk(finish_reason="stop")
+            if raw_blocks:
+                sc.provider_blocks = [raw_blocks[i] for i in sorted(raw_blocks)]
+            yield sc
+
     # -- retryable errors ----------------------------------------------------
 
+    # Computed once at class creation — the retry predicate consults this
+    # per error, and a per-access literal allocates a fresh frozenset each
+    # time.
+    _RETRYABLE_ERROR_NAMES: frozenset[str] = frozenset(
+        {
+            "RateLimitError",
+            "APITimeoutError",
+            "APIConnectionError",
+            "InternalServerError",
+            "APIError",
+            "OverloadedError",
+            # Transport-level: drained stream ended without a stop reason.
+            "IncompleteStreamError",
+        }
+    )
+
     @property
     def retryable_error_names(self) -> frozenset[str]:
-        return frozenset(
-            {
-                "RateLimitError",
-                "APITimeoutError",
-                "APIConnectionError",
-                "InternalServerError",
-                "APIError",
-                "OverloadedError",
-                # Transport-level: drained stream ended without a stop reason.
-                "IncompleteStreamError",
-            }
-        )
+        return self._RETRYABLE_ERROR_NAMES
 
     # -- reasoning extraction ------------------------------------------------
 

turnstone/core/providers/_google.py

@@ -68,9 +68,12 @@ _GOOGLE_DEFAULT = ModelCapabilities(
 class GoogleProvider(OpenAIChatCompletionsProvider):
     """Provider for Google models using the OpenAI-compatible endpoint.
 
-    Overrides message preparation and tool-call extraction to preserve
-    Gemini-specific fields (``thought_signature``) through the round-trip
-    via the ``provider_blocks`` / ``_provider_content`` fidelity lane.
+    Overrides message preparation (``_prepare_messages`` reconstructs the
+    raw tool calls from ``_provider_content``) and registers an
+    ``on_tool_call_delta`` capture with the base stream iterator to
+    preserve Gemini-specific fields (``thought_signature``) through the
+    round-trip via the ``provider_blocks`` / ``_provider_content``
+    fidelity lane.
     """
 
     @property
@@ -139,7 +142,7 @@ class GoogleProvider(OpenAIChatCompletionsProvider):
         stream: Any,
         *,
         finish_reason_optional: bool = False,
-        on_tool_call_delta: Callable[[int, Any], None] | None = None,
+        on_tool_call_delta: Callable[[ToolCallDelta, Any], None] | None = None,
     ) -> Iterator[StreamChunk]:
         """Wrap the base iterator to capture raw tool-call metadata.
 
@@ -161,25 +164,19 @@ class GoogleProvider(OpenAIChatCompletionsProvider):
         """
         raw_tool_calls: dict[int, dict[str, Any]] = {}
 
-        def _capture(slot: int, tc_delta: Any) -> None:
-            fn = tc_delta.function
-            raw_tc = accumulate_tool_call_delta(
-                raw_tool_calls,
-                ToolCallDelta(
-                    index=slot,
-                    id=tc_delta.id or "",
-                    name=(fn.name if fn else None) or "",
-                    arguments_delta=(fn.arguments if fn else None) or "",
-                ),
-            )
+        def _capture(tcd: ToolCallDelta, raw_delta: Any) -> None:
+            # The normalized delta carries the base's slot AND its
+            # id/name/arguments extraction verbatim — the raw lane
+            # accumulates the exact bytes the mirror sees.
+            raw_tc = accumulate_tool_call_delta(raw_tool_calls, tcd)
             # Capture provider-specific extras (e.g. thought_signature)
-            extras = getattr(tc_delta, "__pydantic_extra__", None)
+            extras = getattr(raw_delta, "__pydantic_extra__", None)
             if extras:
                 for k, v in extras.items():
                     if k not in ("index", "id", "type", "function"):
                         raw_tc.setdefault(k, v)
             if on_tool_call_delta is not None:
-                on_tool_call_delta(slot, tc_delta)
+                on_tool_call_delta(tcd, raw_delta)
 
         # Delegate all chunk processing to the base class
         for sc in super()._iter_stream(

turnstone/core/providers/_openai_chat.py

@@ -114,25 +114,35 @@ class ToolCallSlotter:
     - A delta whose id CONTRADICTS the slot's id is a new call; a delta
       whose id MATCHES is the same call, however many times the server
       repeats the name header per fragment.  Ids are authoritative both
-      ways.
-    - An id-less delta with NO name is always an arguments continuation.
-    - An id-less delta announcing a DIFFERENT name than the slot's is a
-      new call (two functions cannot be one call).
-    - An id-less re-announcement of the SAME name splits only when the
-      slot's accumulated arguments are syntactically complete JSON (the
-      previous call is over — this starts the next parallel call).
-      Mid-JSON, the name is a redundant per-fragment header and the
-      delta merges.  With NO arguments accumulated yet: a re-announce
-      that itself carries arguments merges (name-first emission, the
-      arguments are starting now); a bare re-announce splits (the
+      ways — and a slot whose id is KNOWN never splits on an id-less
+      delta at all: on an id-disciplined server new calls arrive with
+      ids, so an id-less fragment (the call's first name announcement,
+      an argument fragment, a redundant header) is always a
+      continuation.  All heuristics below apply only to fully id-less
+      slots.
+    - A delta with NO name is always an arguments continuation.
+    - A name arriving for a slot that has NO name yet is the call's
+      FIRST announcement (e.g. the slot was opened by an args-only
+      delta), never a new call.
+    - A delta announcing a DIFFERENT name than the slot's is a new call
+      (two functions cannot be one call).
+    - A re-announcement of the SAME name splits only when the slot's
+      accumulated arguments are syntactically complete JSON AND the
+      delta itself carries arguments (the next parallel call arriving
+      with its payload).  Mid-JSON, or as a bare trailing delta after
+      complete arguments, the name is a redundant per-fragment header /
+      footer and the delta merges.  With NO arguments accumulated yet:
+      a re-announce that carries arguments merges (name-first emission,
+      the arguments are starting now); a bare re-announce splits (the
       id-less whole-delta shape — two zero-argument parallel calls).
 
     Residual ambiguity, resolved toward the shapes observed in the wild:
     a same-name re-announce on an empty slot that carries arguments is
     read as one call (not a zero-arg call followed by an arg-ful twin),
-    and a trailing bare same-name re-announce after complete arguments
-    is read as a second zero-arg call (not a redundant footer).  Only
-    ids disambiguate those; servers that omit them choose the bet.
+    and a bare same-name re-announce after complete arguments is read
+    as a redundant footer (not a second zero-arg call of the same
+    function).  Only ids disambiguate those; servers that omit them
+    choose the bet.
     """
 
     def __init__(self) -> None:
@@ -152,7 +162,10 @@ class ToolCallSlotter:
             self._slot_ids[slot] = tc_id
         if name:
             self._slot_names[slot] = name
-        if args:
+        # JSON-completeness state is consulted only for fully id-less
+        # slots (``_is_new_call`` short-circuits on a known id), so id'd
+        # slots — the dominant case — skip the per-character scan.
+        if args and slot not in self._slot_ids:
             self._slot_args.setdefault(slot, _ArgsScanner()).feed(args)
         return slot
 
@@ -160,14 +173,28 @@ class ToolCallSlotter:
         slot_id = self._slot_ids.get(slot, "")
         if tc_id:
             return bool(slot_id) and tc_id != slot_id
+        if slot_id:
+            # Id-disciplined slot: new calls on this server arrive with
+            # ids (the id-conflict rule above), so an id-less delta —
+            # the call's first name fragment, an argument fragment, a
+            # redundant header — is always a continuation.
+            return False
         if not name:
             return False
         slot_name = self._slot_names.get(slot, "")
-        if slot_name and name != slot_name:
+        if not slot_name:
+            # First name announcement for a slot opened by an args-only
+            # delta — naming the call, not starting a new one.
+            return False
+        if name != slot_name:
             return True
         scanner = self._slot_args.get(slot)
         if scanner is not None:
-            return scanner.complete
+            # Complete arguments end the call — but only a re-announce
+            # that itself CARRIES arguments starts the next one; a bare
+            # same-name delta after complete arguments is a redundant
+            # footer.
+            return scanner.complete and bool(args)
         return not args
 
 
@@ -315,18 +342,20 @@ class OpenAIChatCompletionsProvider:
         stream: Any,
         *,
         finish_reason_optional: bool = False,
-        on_tool_call_delta: Callable[[int, Any], None] | None = None,
+        on_tool_call_delta: Callable[[ToolCallDelta, Any], None] | None = None,
     ) -> Iterator[StreamChunk]:
         """Convert OpenAI Chat Completions stream chunks to StreamChunks.
 
         *finish_reason_optional* is the model capability of the same name:
         it arms the lax-server finish shim at the end of this generator.
 
-        *on_tool_call_delta* is called with ``(slot, raw_sdk_delta)`` for
-        every tool-call delta, AFTER slot assignment — the seam a subclass
-        uses to capture provider extras (``GoogleProvider``'s
-        ``thought_signature``) keyed by the SAME slot the normalized
-        mirror uses, so the raw and mirror lanes cannot desync.
+        *on_tool_call_delta* is called with ``(normalized, raw_sdk_delta)``
+        for every tool-call delta — the normalized ``ToolCallDelta``
+        carries the slot the base's slotter assigned AND the base's
+        id/name/arguments extraction, so a subclass capturing provider
+        extras (``GoogleProvider``'s ``thought_signature``) accumulates
+        the exact bytes the ``tool_calls`` mirror sees and cannot desync
+        from it on either axis.
         """
         first = True
         annotations: list[Any] = []
@@ -379,8 +408,6 @@ class OpenAIChatCompletionsProvider:
                     name = (fn.name if fn else None) or ""
                     args = (fn.arguments if fn else None) or ""
                     slot = slotter.slot_for(tc_delta.index, tc_id, name, args)
-                    if on_tool_call_delta is not None:
-                        on_tool_call_delta(slot, tc_delta)
                     tcd = ToolCallDelta(index=slot)
                     if tc_id:
                         tcd.id = tc_id
@@ -388,6 +415,8 @@ class OpenAIChatCompletionsProvider:
                         tcd.name = name
                     if args:
                         tcd.arguments_delta = args
+                    if on_tool_call_delta is not None:
+                        on_tool_call_delta(tcd, tc_delta)
                     sc.tool_call_deltas.append(tcd)
                     tool_call_count += 1
 

turnstone/core/providers/_openai_responses.py

@@ -570,17 +570,35 @@ class OpenAIResponsesProvider:
         stream = client.responses.create(**kwargs)
         if cancel_ref is not None:
             cancel_ref.append(stream)
-        return self._iter_stream(stream)
+        caps = capabilities or self.get_capabilities(model)
+        return self._iter_stream(stream, finish_reason_optional=caps.finish_reason_optional)
 
-    def _iter_stream(self, stream: Any) -> Iterator[StreamChunk]:
-        """Convert Responses API stream events to StreamChunks."""
+    def _iter_stream(
+        self, stream: Any, *, finish_reason_optional: bool = False
+    ) -> Iterator[StreamChunk]:
+        """Convert Responses API stream events to StreamChunks.
+
+        *finish_reason_optional* is the model capability of the same name:
+        a lax Responses-compatible server that ends the stream without any
+        terminal event (``response.completed`` / ``response.incomplete``)
+        gets the end-of-generator shim below — the retired non-streaming
+        path needed no terminal event either.
+        """
         first = True
         content_len = 0
+        reasoning_len = 0
         tool_call_count = 0
         last_finish: str | None = None
         completion_tokens: int | None = None
-        # Track tool call indices by call_id for consistent ToolCallDelta.index
+        # Track tool call indices by item_id for consistent ToolCallDelta.index.
+        # ``next_tool_idx`` mints slots (NOT len(dict): duplicate/empty item
+        # ids overwrite their mapping and would collide later slots);
+        # ``last_tool_idx`` routes argument deltas whose item_id was never
+        # announced — a lax server's deltas belong to the call most recently
+        # opened, not hardwired slot 0.
         tool_call_indices: dict[str, int] = {}
+        next_tool_idx = 0
+        last_tool_idx = 0
         # Collect output items for provider_blocks
         provider_blocks: list[dict[str, Any]] = []
         # Collect annotations across text parts
@@ -609,6 +627,7 @@ class OpenAIResponsesProvider:
                 delta_text = getattr(event, "delta", "")
                 if delta_text:
                     sc = StreamChunk(reasoning_delta=delta_text)
+                    reasoning_len += len(delta_text)
                     if first:
                         sc.is_first = True
                         first = False
@@ -637,7 +656,9 @@ class OpenAIResponsesProvider:
                     call_id = getattr(item, "call_id", "")
                     item_id = getattr(item, "id", "")
                     name = getattr(item, "name", "")
-                    idx = len(tool_call_indices)
+                    idx = next_tool_idx
+                    next_tool_idx += 1
+                    last_tool_idx = idx
                     # Index by item_id — argument deltas reference this, not call_id
                     tool_call_indices[item_id] = idx
                     sc = StreamChunk(
@@ -655,7 +676,7 @@ class OpenAIResponsesProvider:
                 item_id = getattr(event, "item_id", "")
                 delta_args = getattr(event, "delta", "")
                 if delta_args:
-                    idx = tool_call_indices.get(item_id, 0)
+                    idx = tool_call_indices.get(item_id, last_tool_idx)
                     yield StreamChunk(
                         tool_call_deltas=[ToolCallDelta(index=idx, arguments_delta=delta_args)]
                     )
@@ -736,8 +757,19 @@ class OpenAIResponsesProvider:
             # 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.
+            # behind a misleading IncompleteStreamError.  AFTER a terminal
+            # event the generation is complete and in hand: a trailing
+            # error frame is teardown noise, and raising would discard the
+            # finished result (the in-band twin of the post-finish
+            # transport-blip tolerance ``drain_stream`` grants).
             if event_type == "error":
+                if last_finish is not None:
+                    log.warning(
+                        "openai.responses.post_terminal_error",
+                        code=getattr(event, "code", "") or "",
+                        message=getattr(event, "message", "") or "",
+                    )
+                    break
                 _raise_responses_failure(
                     getattr(event, "code", "") or "",
                     getattr(event, "message", "Unknown error") or "Unknown error",
@@ -747,16 +779,45 @@ class OpenAIResponsesProvider:
             if event_type == "response.failed":
                 response = getattr(event, "response", None)
                 error = getattr(response, "error", None) if response else None
+                if last_finish is not None:
+                    log.warning(
+                        "openai.responses.post_terminal_error",
+                        code=(getattr(error, "code", "") if error else "") or "",
+                        message=(getattr(error, "message", "") if error else "") or "",
+                    )
+                    break
                 _raise_responses_failure(
                     (getattr(error, "code", "") if error else "") or "",
                     (getattr(error, "message", "") if error else "") or "Unknown error",
                 )
 
+        # Terminal-event-less lax-server tolerance, armed ONLY by the
+        # operator-declared ``finish_reason_optional`` capability: a
+        # stream that ended cleanly after delivering output but never sent
+        # ``response.completed``/``response.incomplete`` is a completed
+        # generation on such a server (the retired non-streaming path
+        # needed no terminal event).  The ``.done``-collected blocks ride
+        # the shimmed finish chunk, exactly as they would the terminal
+        # handler's.  Everywhere else the drain's complete-or-error gate
+        # raises — a missing terminal on an event-disciplined server means
+        # the generation died mid-response.
+        if (
+            finish_reason_optional
+            and last_finish is None
+            and (content_len or reasoning_len or tool_call_count)
+        ):
+            last_finish = "stop"
+            sc = StreamChunk(finish_reason="stop")
+            if provider_blocks:
+                sc.provider_blocks = provider_blocks
+            yield sc
+
         log.debug(
             "openai.responses.response",
             stream=True,
             finish_reason=last_finish,
             content_length=content_len,
+            reasoning_length=reasoning_len,
             tool_call_count=tool_call_count,
             completion_tokens=completion_tokens,
         )

turnstone/core/providers/_protocol.py

@@ -148,10 +148,10 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult:
 
     - A stream that exhausts with NO finish reason raises
       :class:`IncompleteStreamError` (retryable) — every adapter emits one
-      on a healthy stream (a server that genuinely never sends one needs
-      ``finish_reason_optional`` declared in its model capabilities; the
-      chat iterator then shims ``"stop"`` once output arrived), so its
-      absence means the generation died mid-response.  Partial text must
+      on a healthy stream (a server that genuinely never sends a terminal
+      signal needs ``finish_reason_optional`` declared in its model
+      capabilities; the adapter then shims ``"stop"`` once output
+      arrived), so its absence means the generation died mid-response.  Partial text must
       never be handed to a caller that stores it as a complete result (a
       compaction summary, a title).  A transport blip AFTER the finish
       reason keeps the completed result and forfeits only trailing
@@ -358,18 +358,21 @@ class ModelCapabilities:
     # "" = omit (standard reasoning).  There is no gpt-5.6-pro model.
     supports_pro_mode: bool = False
     reasoning_mode: str = ""
-    # Chat-lane lax-server tolerance (operator-declared, model-definition
-    # capabilities JSON): this server never sends ``finish_reason``, so a
+    # Lax-server tolerance (operator-declared, model-definition
+    # capabilities JSON): this server never sends a terminal signal, so a
     # stream that ends CLEANLY after delivering output (content, reasoning,
-    # or tool calls) is a completed generation — the chat iterator shims
-    # ``finish_reason="stop"`` and :func:`drain_stream`'s complete-or-error
-    # gate passes.  Leave False (the default) for every server that
-    # reliably sends finish reasons: there, a clean finish-less end IS a
+    # or tool calls) is a completed generation — the adapter shims a
+    # ``"stop"`` finish and :func:`drain_stream`'s complete-or-error gate
+    # passes.  Leave False (the default) for every server that reliably
+    # terminates its streams: there, a clean signal-less end IS a
     # died-mid-generation stream (worker crashed behind a clean-closing
     # proxy/ASGI layer) and blessing it would store partial text as a
     # complete result.  SSE has no body framing, so the two cases are one
     # wire shape — this flag is the operator asserting which server class
-    # they run.  Honored by the Chat Completions lane only.
+    # they run.  Honored on every drained lane: Chat Completions (no
+    # ``finish_reason`` ever arrived), Anthropic (no ``message_delta``
+    # stop_reason AND no ``message_stop``), Responses (no terminal
+    # ``response.completed``/``response.incomplete`` event).
     finish_reason_optional: bool = False