diff --git a/CHANGELOG.md b/CHANGELOG.md index c1b1d051..0d49c329 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/tests/_session_helpers.py b/tests/_session_helpers.py index 69a9f0e6..c2cd7089 100644 --- a/tests/_session_helpers.py +++ b/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: diff --git a/tests/test_drain_stream.py b/tests/test_drain_stream.py index 63d4c91b..dae5e63c 100644 --- a/tests/test_drain_stream.py +++ b/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: diff --git a/tests/test_providers.py b/tests/test_providers.py index 854f5146..633acb09 100644 --- a/tests/test_providers.py +++ b/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 diff --git a/tests/test_session.py b/tests/test_session.py index bc0c81e5..da05b8d5 100644 --- a/tests/test_session.py +++ b/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) ) diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py index 624027af..4411f9c5 100644 --- a/turnstone/core/providers/_anthropic.py +++ b/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 diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py index 34a4035c..6afd3bdb 100644 --- a/turnstone/core/providers/_openai_chat.py +++ b/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 diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py index 82c151ee..9c023b0c 100644 --- a/turnstone/core/providers/_openai_responses.py +++ b/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) diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 438a8664..76a191f8 100644 --- a/turnstone/core/providers/_protocol.py +++ b/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,