ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(providers): review round 5 — same-id fragment merge, post-finish blip tolerance, chat finish shim

Browse Source 
Correctness:

- ToolCallSlotter's reannounce split is gated to ID-LESS deltas: id
  equality proves the same call, so compat servers that repeat the
  id+name header on every argument fragment merge back into one call
  with valid JSON (round 4's ungated heuristic split them into
  duplicate half-JSON calls — execution-confirmed by the review).  The
  residual id-less repeat-name-per-fragment shape is documented as
  inherently ambiguous; ids are the only disambiguator.
- drain_stream keeps a completed result when the transport blips AFTER
  the finish reason (trailing usage chunk / citation footer window):
  the generation is in hand, so forfeit the trailing metadata instead
  of discarding a fully-delivered verdict or re-paying a compaction.
- The chat iterator shims finish_reason="stop" when a stream ends
  CLEANLY after delivering content or tool calls — the deleted
  non-streaming `or "stop"` default for lax finish-reason-less servers,
  now safe to restore because abrupt deaths surface as
  httpx.TransportError (round 4) rather than clean exhaustion.  This
  supersedes the round-3 keep-the-gate ruling: the httpx catch changed
  the calculus, and the Anthropic/Responses lanes already got their
  marker-based shims.  Empty/reasoning-only streams still fail the
  complete-or-error gate.  Two streaming tests gained the shim chunk.

Dispositions held: o1-era stream-rejecting models (third re-report)
stay a release-note remediation per the earlier ruling.

Cleanup: the two Responses terminal branches collapse into one path
(status derived from the event type when the payload is missing —
also fixes the end-of-stream debug log reporting finish_reason=None
for completed lax streams); the annotations walk is one shared helper
(the two copies had already diverged on None-content guarding);
_raise_responses_failure is annotated NoReturn; scripts/livepass.py
drops the phantom supports_streaming key; test_model_registry's
capture helpers ride scripted_chat_client; _openai_stream_chunk points
at its fake_chat_stream shape-twin for future consolidation.
This commit is contained in:
Patrick Buckley
2026-07-13 14:49:47 -07:00
parent cf7cfe8932
commit 3a28dc2f16
7 changed files with 154 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+6 -5
View File
@@ -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.
scripts/livepass.py
+1 -1
View File
@@ -399,7 +399,7 @@ CONSOLE_TEMPLATE = """<!doctype html>
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,
},
tests/test_model_registry.py
+11 -11
View File
@@ -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
tests/test_providers.py
+70 -3
View File
@@ -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.
turnstone/core/providers/_openai_chat.py
+25 -6
View File
@@ -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()
turnstone/core/providers/_openai_responses.py
+28 -30
View File
@@ -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,
turnstone/core/providers/_protocol.py
+13 -3
View File
@@ -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