diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d49c329..beffad63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,11 +40,12 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. *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. 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 + a current server. Streams that end cleanly after delivering output + complete even when the server never sets a finish reason (each lane + accepts its own terminal marker; the chat lane treats a clean + end-of-stream with content as completion) — only streams that die + mid-generation, deliver nothing, or drop the connection fail, and + those retry before surfacing. 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. diff --git a/scripts/livepass.py b/scripts/livepass.py index 50ed5507..d6348467 100755 --- a/scripts/livepass.py +++ b/scripts/livepass.py @@ -399,7 +399,7 @@ CONSOLE_TEMPLATE = """ known: true, capabilities: { context_window: 200000, supports_tools: true, - supports_streaming: true, supports_vision: true, + supports_vision: true, supports_web_search: true, supports_temperature: true, supports_effort: true, }, diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py index 20ace732..600659e5 100644 --- a/tests/test_model_registry.py +++ b/tests/test_model_registry.py @@ -8,7 +8,7 @@ from unittest.mock import MagicMock, patch import pytest -from tests._session_helpers import fake_chat_stream +from tests._session_helpers import scripted_chat_client from turnstone.core.model_registry import ( ModelConfig, ModelRegistry, @@ -1300,13 +1300,8 @@ class TestSessionAgentModel: ) session = _make_session(registry=reg, model_alias="main") - # Mock the API to capture what model was used - captured_model = None - - def fake_create(**kwargs: Any) -> Any: - nonlocal captured_model - captured_model = kwargs.get("model") - return fake_chat_stream(content="done") + # Scripted client records kwargs; read the model off its calls. + fake_create = scripted_chat_client({"content": "done"}) # Get the agent client from the registry and patch it agent_client = reg.get_client("agent") @@ -1317,16 +1312,21 @@ class TestSessionAgentModel: Turn.user("Do something."), ] session._run_agent(agent_msgs) - assert captured_model == "agent-model" + assert fake_create.calls[-1].get("model") == "agent-model" @staticmethod def _capture_on(client: Any) -> dict[str, Any]: - """Patch *client* (registry-resolved or session.client) to capture kwargs.""" + """Patch *client* (registry-resolved or session.client) to capture kwargs. + + Rides the shared scripted client; the returned dict mirrors the + LAST call's kwargs (existing reader contract). + """ captured: dict[str, Any] = {} + scripted = scripted_chat_client({"content": "done"}) def fake_create(**kwargs: Any) -> Any: captured.update(kwargs) - return fake_chat_stream(content="done") + return scripted(**kwargs) client.chat.completions.create = fake_create return captured diff --git a/tests/test_providers.py b/tests/test_providers.py index 648b9f39..7922cf8f 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -48,7 +48,12 @@ def _openai_stream_chunk( usage: MagicMock | None = None, empty_choices: bool = False, ) -> MagicMock: - """Build a mock OpenAI streaming chunk.""" + """Build a mock OpenAI streaming chunk. + + Shape twin of ``tests/_session_helpers.fake_chat_stream`` (which + builds whole scripted streams on SimpleNamespace); consolidate onto + one fake when either next changes shape. + """ chunk = MagicMock() if empty_choices: chunk.choices = [] @@ -569,9 +574,12 @@ class TestOpenAIProvider: messages=[{"role": "user", "content": "hi"}], ) ) - assert len(results) == 2 + # Third chunk: the finish-reason-less shim stamps the clean end + # of a content-bearing stream as "stop". + assert len(results) == 3 assert results[0].content_delta == "Hello" assert results[1].content_delta == " world" + assert results[2].finish_reason == "stop" def test_streaming_reasoning(self) -> None: chunks = [ @@ -612,11 +620,14 @@ class TestOpenAIProvider: messages=[{"role": "user", "content": "read a file"}], ) ) - assert len(results) == 3 + # Fourth chunk: the finish-reason-less shim (tool calls count as + # delivered output). + assert len(results) == 4 assert results[0].tool_call_deltas[0].id == "call_1" assert results[0].tool_call_deltas[0].name == "read_file" assert results[1].tool_call_deltas[0].arguments_delta == '{"path":' assert results[2].tool_call_deltas[0].arguments_delta == '"foo.py"}' + assert results[3].finish_reason == "stop" def test_streaming_usage(self) -> None: usage = MagicMock() @@ -804,6 +815,62 @@ class TestOpenAIProvider: assert result.tool_calls[0]["function"]["arguments"] == '{"a": 1}' assert result.tool_calls[1]["function"]["arguments"] == '{"b": 2}' + def test_repeated_id_and_name_header_fragments_stay_one_call(self) -> None: + # Some compat servers repeat the full id+name header on EVERY + # argument fragment. Id equality proves same call — the + # reannounce split applies only to ID-LESS deltas, so this shape + # merges into one call with valid arguments (round-5 regression: + # the ungated heuristic split it into duplicate half-JSON calls). + result = self._drain_chunks( + [ + _openai_stream_chunk( + tool_calls=[ + _openai_tool_call_delta( + index=0, tc_id="call_A", name="read_file", arguments='{"path": ' + ) + ] + ), + _openai_stream_chunk( + tool_calls=[ + _openai_tool_call_delta( + index=0, tc_id="call_A", name="read_file", arguments='"/tmp/x"}' + ) + ] + ), + _openai_stream_chunk(finish_reason="tool_calls"), + ] + ) + assert len(result.tool_calls) == 1 + assert result.tool_calls[0]["function"]["arguments"] == '{"path": "/tmp/x"}' + + def test_finishless_stream_with_content_completes_as_stop(self) -> None: + # Lax-server tolerance (the deleted non-streaming `or "stop"` + # default): a stream that ends CLEANLY after delivering content — + # abrupt deaths raise httpx errors instead — is a completed + # generation even if no chunk ever carried finish_reason. + result = self._drain_chunks( + [ + _openai_stream_chunk(content="complete answer"), + ] + ) + assert result.content == "complete answer" + assert result.finish_reason == "stop" + + def test_finishless_stream_with_no_output_still_raises(self) -> None: + # The shim is content-gated: an empty clean-close stream (dead + # generation, zero-chunk fakes) still hits the drain's + # complete-or-error gate. + from turnstone.core.providers import IncompleteStreamError + + client = MagicMock() + client.chat.completions.create.return_value = [] + with pytest.raises(IncompleteStreamError): + drain_stream( + self.provider.create_streaming( + client=client, model="m", messages=[{"role": "user", "content": "x"}] + ) + ) + def test_fragmented_single_call_does_not_split(self) -> None: # The normal well-behaved shape — name announced once, arguments # streamed in fragments — must stay ONE call. diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py index 902498a1..81dc1cf6 100644 --- a/turnstone/core/providers/_openai_chat.py +++ b/turnstone/core/providers/_openai_chat.py @@ -63,11 +63,16 @@ class ToolCallSlotter: the normalized mirror can never desync from the raw ``provider_blocks`` lane). Index-degenerate compat servers (historical vLLM/llama.cpp builds) emit every parallel call at index 0 — a delta opens a NEW slot - when its id contradicts the slot's id, or when it announces a name for - a slot that already accumulated arguments (the id-less whole-delta - shape: name+arguments per call, so a second announcement after - arguments is a second call). Id-less argument fragments keep - following their index's current slot. + when its id contradicts the slot's id, or when an ID-LESS delta + announces a name for a slot that already accumulated arguments (the + id-less whole-delta shape: name+arguments per call, so a second + announcement after arguments is a second call). A delta whose id + MATCHES the slot's is always the same call, however many times the + server repeats the name header per fragment. Residual ambiguity: + an id-less server that repeats the name on every argument fragment is + indistinguishable from the whole-delta shape and splits wrongly — ids + are the only disambiguator, and the whole-delta emission is the shape + observed in the wild. """ def __init__(self) -> None: @@ -84,7 +89,7 @@ class ToolCallSlotter: and self._slot_ids.get(slot, "") and self._slot_ids[slot] != tc_id ) - reannounce = slot is not None and has_name and slot in self._slot_has_args + reannounce = slot is not None and not tc_id and has_name and slot in self._slot_has_args if slot is None or id_conflict or reannounce: slot = self._next_slot self._next_slot += 1 @@ -322,6 +327,20 @@ class OpenAIChatCompletionsProvider: completion_tokens=completion_tokens, ) + # Finish-reason-less compat tolerance (the deleted non-streaming + # path's `finish_reason or "stop"` default): a stream that ended + # CLEANLY — the SDK ends iteration on [DONE]; an abrupt connection + # death raises httpx.TransportError out of this generator — after + # delivering content is a completed generation on a lax server + # that never sets finish_reason. Emit the finish BEFORE the + # citation footer so the drain folds the footer as trailing info. + # A clean-exhaustion stream with NO output still yields nothing, + # so the drain's complete-or-error gate keeps catching dead/empty + # streams. + if last_finish_reason is None and (content_len or tool_call_count): + last_finish_reason = "stop" + yield StreamChunk(finish_reason="stop") + # Emit accumulated citations as a final info chunk if annotations: citation_text = format_citations("", annotations).strip() diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py index 850c8bba..989cb238 100644 --- a/turnstone/core/providers/_openai_responses.py +++ b/turnstone/core/providers/_openai_responses.py @@ -8,7 +8,7 @@ of the Chat Completions endpoint. from __future__ import annotations import json -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, NoReturn if TYPE_CHECKING: from collections.abc import Callable, Iterator @@ -48,7 +48,20 @@ log = structlog.get_logger(__name__) _TRANSIENT_FAILURE_CODES = frozenset({"server_error", "rate_limit_exceeded"}) -def _raise_responses_failure(error_code: str, error_msg: str) -> None: +def _extend_message_annotations(item: Any, annotations: list[Any]) -> None: + """Collect url_citation annotations off a message output item's text + parts — ONE walk shared by the ``output_item.done`` handler and the + terminal-payload rebuild so the two cannot drift (``content`` may be + ``None`` on partial items).""" + if getattr(item, "type", "") != "message": + return + for content_part in getattr(item, "content", None) or []: + part_anns = getattr(content_part, "annotations", None) + if part_anns: + annotations.extend(part_anns) + + +def _raise_responses_failure(error_code: str, error_msg: str) -> NoReturn: """One classification ladder for BOTH in-band failure shapes (`error` events and ``response.failed``) — a one-sided edit would make the same API failure retryable through one event type and fatal through the @@ -663,12 +676,7 @@ class OpenAIResponsesProvider: item_dict = item.model_dump() if hasattr(item, "model_dump") else {} if item_dict: provider_blocks.append(item_dict) - # Collect annotations from completed text parts - if getattr(item, "type", "") == "message": - for content_part in getattr(item, "content", []): - part_anns = getattr(content_part, "annotations", None) - if part_anns: - annotations.extend(part_anns) + _extend_message_annotations(item, annotations) continue # -- terminal response event -- @@ -678,23 +686,17 @@ 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, keeping the blocks - # already collected from output_item.done events (they - # came from the stream, not the missing payload); only - # usage is genuinely unavailable. - sc = StreamChunk( - finish_reason="stop" if event_type == "response.completed" else "length" - ) - if provider_blocks: - sc.provider_blocks = provider_blocks - yield sc - continue - status = getattr(response, "status", "completed") + # A terminal event without its payload (lax compat server) + # is still a terminal signal: derive the finish reason from + # the event type, keep the blocks already collected from + # output_item.done events — only usage (and the rebuild + # below) genuinely needs the payload. + if response is not None: + status = getattr(response, "status", "") + else: + status = "completed" if event_type == "response.completed" else "incomplete" last_finish = "stop" if status == "completed" else "length" - usage = extract_usage(getattr(response, "usage", None)) + usage = extract_usage(getattr(response, "usage", None)) if response else None if usage: completion_tokens = usage.completion_tokens # Prefer the terminal response's own output items over the @@ -705,18 +707,14 @@ class OpenAIResponsesProvider: # turn's replay a 400. Its annotations were likewise # never collected — walk them here (format_citations # dedupes by URL, so re-seeing .done'd items is harmless). - out_items = getattr(response, "output", None) or [] + out_items = (getattr(response, "output", None) or []) if response else [] final_items = [ item.model_dump() for item in out_items if hasattr(item, "model_dump") ] if final_items: provider_blocks = final_items for item in out_items: - if getattr(item, "type", "") == "message": - for content_part in getattr(item, "content", []) or []: - part_anns = getattr(content_part, "annotations", None) - if part_anns: - annotations.extend(part_anns) + _extend_message_annotations(item, annotations) sc = StreamChunk( finish_reason=last_finish, usage=usage, diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 1213796f..02372f4c 100644 --- a/turnstone/core/providers/_protocol.py +++ b/turnstone/core/providers/_protocol.py @@ -120,9 +120,13 @@ 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, 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). + on a healthy stream (the chat iterator shims ``"stop"`` for lax + finish-reason-less servers once content 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 + metadata. - ``usage`` merges via :func:`merge_usage` — Anthropic splits prompt and completion tokens across separate events. - Tool calls accumulate by ``ToolCallDelta.index``: ``id``/``name`` @@ -166,6 +170,12 @@ def drain_stream(chunks: Iterator[StreamChunk]) -> CompletionResult: except StopIteration: break except httpx.TransportError as exc: + if finish_reason is not None: + # The generation already completed (finish reason in hand); + # the blip only cost trailing metadata — a usage-only chunk + # or the citation footer. Keep the complete result rather + # than discarding it for a retry that re-pays the tokens. + break raise IncompleteStreamError( f"stream transport failed mid-response ({type(exc).__name__}: {exc})" ) from exc