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 9 — orphan/harvest collision, shared shim gate, retired-id rationale

Browse Source 
Correctness:
- Responses: orphan argument deltas (streamed without any
  output_item.added) now count as a streamed tool-call signal, so the
  terminal harvest stands down instead of re-emitting the same call
  onto the same slot — the reproduced collision concatenated the
  arguments JSON into an unparseable double copy.

Cleanup / documentation:
- finish_shim_due in _protocol is THE gate for the lax-server finish
  shim — one predicate (and one definition of 'delivered output') for
  all three adapter families, so the same capability flag cannot
  acquire per-family completion semantics.
- The Responses error/response.failed branches share one failure tail
  (only code/message extraction differs) — the same server failure can
  never become retryable through one event type and fatal through the
  other, pre- or post-terminal.
- _format_refusal pins the refusal rendering the streamed event and
  the terminal harvest both use.
- The capability-table floor comment and CHANGELOG Removed entry now
  state the real rationale: OpenAI has RETIRED the pruned ids from the
  API — the rows described unreachable contracts, not unpopular ones.
- CHANGELOG names the stream-entitlement break class (verified-org
  streaming, pre-stream_options gateway api-versions) with its
  serving-side remediation; deliberately no non-streaming fallback.
- docs/architecture.md retry section describes the collapsed
  transport: the two stacked retry ladders, IncompleteStreamError /
  ResponsesStreamFailedError retryability, finish_reason_optional
  remediation; stale non-streaming mentions updated (+ puml).
- Anthropic whole-block emission carries its residual hybrid-gateway
  bet as an explicit comment.

Held on standing rulings: post-finish usage forfeiture (keep result +
warn, rounds 4/8), session merge_usage twin and StreamAbortRef twin
(#832), stream_options wire delta (round 2, caveat now names Azure).
This commit is contained in:
Patrick Buckley
2026-07-13 22:19:28 -07:00
parent 49d33d8594
commit 747177a76c
9 changed files with 165 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+15 -8
View File
@@ -29,7 +29,13 @@ 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. These lanes are also complete-or-error now: a stream
on a dead call. Because every call now streams, an alias pointed at a
model or org that cannot stream (OpenAI's verified-org streaming
entitlement, a gateway api-version predating `stream_options` — e.g.
older Azure OpenAI deployments) fails at request time where 1.7's
non-streaming single-shot call succeeded; remediation is on the
serving side (verify the org, bump the api-version/gateway) — there is
deliberately no per-model non-streaming fallback left to configure. 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
@@ -142,13 +148,14 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
table.** `o1`, `o1-mini`, `o3`, `o3-mini`, `o3-pro`, `o4-mini`,
`gpt-5`, `gpt-5-mini`, `gpt-5-nano`, `gpt-5-pro`, `gpt-5.1`,
`gpt-5.1-codex-max`, `gpt-5.2`, `gpt-5.2-pro`, and `gpt-5.3` no longer
have built-in capability rows — these models are effectively unused in
the field. The table floor is now `gpt-5.4`; the search-api and
audio/STT/TTS rows are unchanged. An alias still pinning one of the
removed ids resolves to the generic commercial defaults (temperature
sent, no declared reasoning-effort vocabulary, 200K window), which
those models may reject — declare the contract on the model
definition's capabilities JSON if you must stay on one, or move to a
have built-in capability rows — OpenAI has retired these model ids
from the API, so the rows described contracts no request can reach
anymore. The table floor is now `gpt-5.4`; the search-api and
audio/STT/TTS rows are unchanged. An alias still pinning a retired id
fails at OpenAI itself; any other unlisted commercial id resolves to
the generic commercial defaults (temperature sent, no declared
reasoning-effort vocabulary, 200K window) — declare the contract on
the model definition's capabilities JSON if you run one, or move to a
current model.
### Fixed
docs/architecture.md
+25 -15
View File
@@ -662,7 +662,7 @@ display). Automatic prompt caching is enabled via top-level `cache_control:
cacheable block and advances it as conversations grow (90% input cost
reduction on cache hits, 1.25x write on first turn). Cache metrics
(`cache_creation_input_tokens`, `cache_read_input_tokens`) are extracted from
both streaming and non-streaming responses. The `anthropic` SDK is a core
the stream's usage events. The `anthropic` SDK is a core
dependency — the Anthropic provider is first-class alongside OpenAI.
**GoogleProvider** (`_google.py`): extends `OpenAIChatCompletionsProvider` for
@@ -1149,20 +1149,30 @@ Named (aliased) workstreams are never age-pruned. Configure with
### API Retry
`ChatSession._create_stream_with_retry()` (streaming path) and the agent
`_api_call()` (non-streaming) both use the same retry pattern:
Every model call streams (#831); retry lives at two stacked layers:
- **Retries**: 4 total attempts (1 initial + 3 retries, `_MAX_RETRIES = 3`)
- **Backoff**: exponential, base 1 second (`delay = 1s * 2^attempt`)
- **Retryable errors**: `RateLimitError`, `APITimeoutError`,
`APIConnectionError`, `InternalServerError`, `ServiceUnavailableError`,
`APIError` (matched by class name to avoid importing backend-specific
exception hierarchies)
- On retry: `ui.on_info()` notification
- On final failure: exception propagates
`_compact_messages()` also wraps its non-streaming API call in the same
retry loop.
- **Caller ladders** — `ChatSession._create_stream_with_retry()` (chat
loop) and the agent `_api_call()` (drained via `model_turn`) use the
same pattern: 4 total attempts (1 initial + 3 retries,
`_MAX_RETRIES = 3`), exponential backoff base 1 second
(`delay = 1s * 2^attempt`), `ui.on_info()` on retry, exception
propagates on final failure. `_compact_messages()` wraps its drained
call in the same loop.
- **`model_turn`'s drain ladder** — inside every single-shot call,
mid-stream deaths (errors raised while draining, e.g.
`IncompleteStreamError`) are re-issued up to 2 more times with a
0.5s-base exponential backoff (±50% jitter); request-time failures
keep the SDK's own retry policy. The two ladders stack
multiplicatively on transient-shaped failures.
- **Retryable errors** are matched by class name against each
provider's `retryable_error_names` (avoids importing
backend-specific exception hierarchies): `RateLimitError`,
`APITimeoutError`, `APIConnectionError`, `InternalServerError`,
`ServiceUnavailableError`, `APIError`, plus the drained-transport
errors `IncompleteStreamError` (stream ended with no terminal
signal — for servers that never send one, declare
`finish_reason_optional` in the model's capabilities JSON) and
`ResponsesStreamFailedError` (transient in-band Responses failure).
### Finish Reason Handling
@@ -1175,7 +1185,7 @@ retry loop.
blocked.
Agent sub-sessions (`_run_agent()`) check `finish_reason` on each
non-streaming response and stop the agent early on `"length"` or
drained turn and stop the agent early on `"length"` or
`"content_filter"`.
`_compact_messages()` checks `finish_reason` on the compaction response and
docs/diagrams/03-core-engine-classes.puml
+1 -1
View File
@@ -176,7 +176,7 @@ class "HeadlessSession" as HeadlessSession {
+ send_headless(input, max_turns, ...)
- _override_system_prompt(content)
--
eval.py: non-streaming,
eval.py: drained single-shot turns,
records all tool calls
}
tests/test_providers.py
+30
View File
@@ -5760,6 +5760,36 @@ class TestResponsesDrainedStream:
assert by_name["beta"] == '{"x": 1}'
assert by_name["alpha"] == ""
def test_orphan_deltas_do_not_collide_with_terminal_harvest(self) -> None:
# Reproduced round-9 regression: argument deltas streamed without
# any output_item.added announcement accumulate at slot 0, and the
# terminal harvest (gated on "no tool calls streamed") re-emitted
# the same call onto the same slot — concatenating the arguments
# into '{"x": 1}{"x": 1}'. Orphan deltas ARE a streamed tool-call
# signal, so the harvest must stand down.
terminal_item = SimpleNamespace(type="function_call")
terminal_item.model_dump = lambda **_kw: { # type: ignore[method-assign]
"type": "function_call",
"call_id": "call_1",
"name": "do_thing",
"arguments": '{"x": 1}',
}
events = [
SimpleNamespace(
type="response.function_call_arguments.delta",
item_id="never_announced",
delta='{"x": 1}',
),
SimpleNamespace(
type="response.completed",
response=SimpleNamespace(status="completed", usage=None, output=[terminal_item]),
),
]
result = self._drain(events)
assert result.tool_calls is not None
assert len(result.tool_calls) == 1
assert result.tool_calls[0]["function"]["arguments"] == '{"x": 1}'
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
turnstone/core/providers/_anthropic.py
+14 -1
View File
@@ -20,6 +20,7 @@ from turnstone.core.providers._protocol import (
UsageInfo,
_join_reasoning_with_cap,
_lookup_capabilities,
finish_shim_due,
merge_reasoning_template_kwargs,
snap_reasoning_effort,
)
@@ -969,6 +970,14 @@ class AnthropicProvider:
# every drained lane return a clean-looking empty result
# where the retired non-streaming path (SDK
# get_final_message accumulation) returned the content.
# Residual bet, documented: a HYBRID gateway that sends a
# populated start AND re-streams the same content as
# deltas would double-count. No known server does this
# (it would double on the SDK's own accumulators too), and
# the two attested classes — real API (empty starts),
# whole-block gateways (no deltas) — are both handled;
# disambiguating would mean buffering every start block
# until its first delta or block_stop.
if block.type == "text":
# Separate consecutive text blocks the way the Messages
# API's non-streaming shape reads when joined — without
@@ -1163,7 +1172,11 @@ class AnthropicProvider:
# 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:
if finish_shim_due(
finish_reason_optional=finish_reason_optional,
finish_seen=emitted_finish,
delivered_output=delivered_output,
):
sc = StreamChunk(finish_reason="stop")
_attach_terminal_blocks(sc)
yield sc
turnstone/core/providers/_openai_chat.py
+5 -4
View File
@@ -27,6 +27,7 @@ from turnstone.core.providers._protocol import (
StreamChunk,
ToolCallDelta,
_join_reasoning_with_cap,
finish_shim_due,
merge_reasoning_template_kwargs,
)
from turnstone.core.trajectory import materialize_attachments
@@ -451,10 +452,10 @@ class OpenAIChatCompletionsProvider:
# the drain folds the footer as trailing info. A clean-exhaustion
# stream with NO output still yields nothing, so dead/empty
# streams keep failing the gate even when the shim is armed.
if (
finish_reason_optional
and last_finish_reason is None
and (content_len or reasoning_len or tool_call_count)
if finish_shim_due(
finish_reason_optional=finish_reason_optional,
finish_seen=last_finish_reason is not None,
delivered_output=bool(content_len or reasoning_len or tool_call_count),
):
last_finish_reason = "stop"
yield StreamChunk(finish_reason="stop")
turnstone/core/providers/_openai_common.py
+8 -4
View File
@@ -29,10 +29,14 @@ log = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Table floor: gpt-5.4 (2026-07 pruning) — the o-series and pre-5.4 GPT-5
# rows were dropped as unused in the field. A legacy commercial id now
# resolves to OPENAI_DEFAULT (generic caps: temperature sent, no declared
# effort vocabulary); anyone still pinning one declares the contract on
# the model definition's capabilities JSON or moves to a current model.
# rows were dropped because OpenAI has RETIRED those model ids from the
# API (they are no longer served, not merely unpopular), so no live
# deployment can be calling them and the rows described unreachable
# contracts. A request for a retired id fails at OpenAI regardless of
# what this table says. Any OTHER unlisted commercial id resolves to
# OPENAI_DEFAULT (generic caps: temperature sent, no declared effort
# vocabulary); an operator pinning one declares the contract on the
# model definition's capabilities JSON or moves to a current model.
OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
# GPT-5.4 — 1M context window, native tool search
"gpt-5.4": ModelCapabilities(
turnstone/core/providers/_openai_responses.py
+48 -42
View File
@@ -35,6 +35,7 @@ from turnstone.core.providers._protocol import (
StreamChunk,
ToolCallDelta,
_join_reasoning_with_cap,
finish_shim_due,
resolve_reasoning_effort,
)
from turnstone.core.trajectory import materialize_attachments
@@ -48,6 +49,13 @@ log = structlog.get_logger(__name__)
_TRANSIENT_FAILURE_CODES = frozenset({"server_error", "rate_limit_exceeded"})
def _format_refusal(text: str) -> str:
"""Render a refusal part as visible content — ONE format for both the
streamed ``response.refusal.done`` event and the terminal-payload
harvest, so drained text cannot differ by which path carried it."""
return f"[Refused: {text}]"
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
@@ -599,6 +607,7 @@ class OpenAIResponsesProvider:
tool_call_indices: dict[str, int] = {}
next_tool_idx = 0
last_tool_idx = 0
orphan_args_seen = False
# Collect output items for provider_blocks
provider_blocks: list[dict[str, Any]] = []
# Collect annotations across text parts
@@ -641,7 +650,7 @@ class OpenAIResponsesProvider:
# double-emit.
if event_type == "response.refusal.done":
refusal_text = getattr(event, "refusal", "")
sc = StreamChunk(content_delta=f"[Refused: {refusal_text}]")
sc = StreamChunk(content_delta=_format_refusal(refusal_text))
content_len += len(sc.content_delta)
if first:
sc.is_first = True
@@ -676,6 +685,13 @@ class OpenAIResponsesProvider:
item_id = getattr(event, "item_id", "")
delta_args = getattr(event, "delta", "")
if delta_args:
if item_id not in tool_call_indices:
# Orphan deltas ARE a streamed tool-call signal:
# without this flag the terminal harvest (gated on
# "no tool calls streamed") re-emits the same call
# onto the same slot and the arguments JSON
# duplicates.
orphan_args_seen = True
idx = tool_call_indices.get(item_id, last_tool_idx)
yield StreamChunk(
tool_call_deltas=[ToolCallDelta(index=idx, arguments_delta=delta_args)]
@@ -764,7 +780,7 @@ class OpenAIResponsesProvider:
if part.get("type") == "output_text" and part.get("text"):
parts_text.append(part["text"])
elif part.get("type") == "refusal" and part.get("refusal"):
parts_text.append(f"[Refused: {part['refusal']}]")
parts_text.append(_format_refusal(part["refusal"]))
harvested = "".join(parts_text)
if harvested:
content_len = len(harvested)
@@ -773,7 +789,7 @@ class OpenAIResponsesProvider:
hc.is_first = True
first = False
yield hc
if tool_call_count == 0:
if tool_call_count == 0 and not orphan_args_seen:
for block in provider_blocks:
if not (isinstance(block, dict) and block.get("type") == "function_call"):
continue
@@ -803,43 +819,33 @@ 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. 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":
# -- in-band failure events --
# ``error`` (ResponseErrorEvent): the SDK YIELDS these 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. ``response.failed`` carries the same
# failure nested in its response payload. ONE tail for both
# shapes (only the code/message extraction differs), so the
# same server failure can never become retryable through one
# event type and fatal through the other: BEFORE a terminal
# event the failure raises; AFTER one the generation is
# complete and in hand — a trailing failure frame is teardown
# noise (the in-band twin of the post-finish transport-blip
# tolerance ``drain_stream`` grants), logged and dropped.
if event_type in ("error", "response.failed"):
if event_type == "error":
code = getattr(event, "code", "") or ""
message = getattr(event, "message", "") or ""
else:
response = getattr(event, "response", None)
error = getattr(response, "error", None) if response else None
code = (getattr(error, "code", "") if error else "") or ""
message = (getattr(error, "message", "") if error else "") or ""
if last_finish is not None:
log.warning(
"openai.responses.post_terminal_error",
code=getattr(event, "code", "") or "",
message=getattr(event, "message", "") or "",
)
log.warning("openai.responses.post_terminal_error", code=code, message=message)
break
_raise_responses_failure(
getattr(event, "code", "") or "",
getattr(event, "message", "Unknown error") or "Unknown error",
)
# -- error --
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",
)
_raise_responses_failure(code, message or "Unknown error")
# Terminal-event-less lax-server tolerance, armed ONLY by the
# operator-declared ``finish_reason_optional`` capability: a
@@ -851,10 +857,10 @@ class OpenAIResponsesProvider:
# 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)
if finish_shim_due(
finish_reason_optional=finish_reason_optional,
finish_seen=last_finish is not None,
delivered_output=bool(content_len or reasoning_len or tool_call_count),
):
last_finish = "stop"
sc = StreamChunk(finish_reason="stop")
turnstone/core/providers/_protocol.py
+19
View File
@@ -109,6 +109,25 @@ def merge_usage(acc: UsageInfo | None, new: UsageInfo) -> UsageInfo:
)
def finish_shim_due(
*, finish_reason_optional: bool, finish_seen: bool, delivered_output: bool
) -> bool:
"""THE gate for the lax-server finish shim, shared by every adapter.
A stream that ended cleanly without any terminal signal is shimmed to
``"stop"`` only when ALL of: the operator declared
``finish_reason_optional`` on the model (this server never sends
terminal signals), no finish was seen (the shim never overrides a
real signal), and output was DELIVERED — content, reasoning, or tool
calls; ``info_delta``/usage do not count (status pings are not a
generation). Dead/empty streams keep failing the drain's
complete-or-error gate even when the flag is armed. One predicate so
the same capability flag cannot acquire different completion
semantics per provider family.
"""
return finish_reason_optional and not finish_seen and delivered_output
def accumulate_tool_call_delta(
acc: dict[int, dict[str, Any]], tcd: ToolCallDelta
) -> dict[str, Any]: