diff --git a/CHANGELOG.md b/CHANGELOG.md index 0dad06fe..3c6cdd9e 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/tests/_session_helpers.py b/tests/_session_helpers.py index c2cd7089..b1a0a0a5 100644 --- a/tests/_session_helpers.py +++ b/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) diff --git a/tests/test_model_turn.py b/tests/test_model_turn.py index a5663f39..5c19c327 100644 --- a/tests/test_model_turn.py +++ b/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") diff --git a/tests/test_providers.py b/tests/test_providers.py index 34d0129f..06465dfd 100644 --- a/tests/test_providers.py +++ b/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 diff --git a/turnstone/core/model_turn.py b/turnstone/core/model_turn.py index ae726f42..669745a1 100644 --- a/turnstone/core/model_turn.py +++ b/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 diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py index 4411f9c5..0a655c52 100644 --- a/turnstone/core/providers/_anthropic.py +++ b/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 ------------------------------------------------ diff --git a/turnstone/core/providers/_google.py b/turnstone/core/providers/_google.py index 872b81f8..519c9cf3 100644 --- a/turnstone/core/providers/_google.py +++ b/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( diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py index 4a8ec249..f2a596c6 100644 --- a/turnstone/core/providers/_openai_chat.py +++ b/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 diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py index 3911b246..94b3a508 100644 --- a/turnstone/core/providers/_openai_responses.py +++ b/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, ) diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 4acef5a2..1871b07e 100644 --- a/turnstone/core/providers/_protocol.py +++ b/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