From 747177a76c6051d0023e706bf1164ce051fabb68 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 13 Jul 2026 22:19:28 -0700 Subject: [PATCH] =?UTF-8?q?fix(providers):=20review=20round=209=20?= =?UTF-8?q?=E2=80=94=20orphan/harvest=20collision,=20shared=20shim=20gate,?= =?UTF-8?q?=20retired-id=20rationale?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- CHANGELOG.md | 23 +++-- docs/architecture.md | 40 +++++---- docs/diagrams/03-core-engine-classes.puml | 2 +- tests/test_providers.py | 30 +++++++ turnstone/core/providers/_anthropic.py | 15 +++- turnstone/core/providers/_openai_chat.py | 9 +- turnstone/core/providers/_openai_common.py | 12 ++- turnstone/core/providers/_openai_responses.py | 90 ++++++++++--------- turnstone/core/providers/_protocol.py | 19 ++++ 9 files changed, 165 insertions(+), 75 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c6cdd9e..c3374c09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/docs/architecture.md b/docs/architecture.md index 54d0816b..83c8e2d8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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 diff --git a/docs/diagrams/03-core-engine-classes.puml b/docs/diagrams/03-core-engine-classes.puml index d7cdee64..2f88512c 100644 --- a/docs/diagrams/03-core-engine-classes.puml +++ b/docs/diagrams/03-core-engine-classes.puml @@ -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 } diff --git a/tests/test_providers.py b/tests/test_providers.py index fe93630d..b8fd7376 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -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 diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py index 87e5366f..1f6b39cf 100644 --- a/turnstone/core/providers/_anthropic.py +++ b/turnstone/core/providers/_anthropic.py @@ -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 diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py index f2a596c6..d3d0590a 100644 --- a/turnstone/core/providers/_openai_chat.py +++ b/turnstone/core/providers/_openai_chat.py @@ -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") diff --git a/turnstone/core/providers/_openai_common.py b/turnstone/core/providers/_openai_common.py index b56b89ba..1c1a112a 100644 --- a/turnstone/core/providers/_openai_common.py +++ b/turnstone/core/providers/_openai_common.py @@ -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( diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py index c8657628..d973beb1 100644 --- a/turnstone/core/providers/_openai_responses.py +++ b/turnstone/core/providers/_openai_responses.py @@ -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") diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 13fa9879..0c2d89a2 100644 --- a/turnstone/core/providers/_protocol.py +++ b/turnstone/core/providers/_protocol.py @@ -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]: