diff --git a/agent_core/errors.py b/agent_core/errors.py index 62b3e06..5a8ca19 100644 --- a/agent_core/errors.py +++ b/agent_core/errors.py @@ -108,6 +108,77 @@ def __init__(self, *, chunks_seen: int, elapsed_s: float) -> None: ) +class LLMTruncatedStream(LLMError): + """An Anthropic stream ended without its protocol terminator. + + The Messages streaming protocol always closes with ``message_delta`` + (stop_reason + output usage) followed by ``message_stop``. A stream that + ends before ``message_stop`` was cut off, whatever it carried so far, and + the client cannot tell that from a deliberate stop: the SDK raises only on + an SSE ``error`` event or a broken chunked body, and a gateway that loses + its upstream mid-generation can do neither and simply finish the response + body. What comes back is well-formed and short — thinking, no text, no tool + call — and the loop reads it as "the model chose to stop" and ends the run. + + Measured on ApodexHarness's 2026-10-07 GDPval batch over llm-hub: 10 of 15 + finished trials ended this way, in three bursts where several unrelated + in-flight streams were cut in the same second (02:42:20, 03:00:35 and + 03:07:46 UTC). Every one still carried ``message_start`` usage, so the + empty-completion guard (which requires *no* usage) let it through. + + The wording matches ``_TRUNCATED_STREAM_PATTERNS``: the turn is resampled + on the same key, and an active fallback chain advances instead. The + carried fields are the stream's end record, so the cause is visible in the + retry log and attempt telemetry instead of only in a missing deliverable. + """ + + def __init__( + self, + *, + last_event: str, + events_seen: int, + saw_message_delta: bool, + block_types: list[str], + elapsed_s: float, + ) -> None: + self.last_event = last_event + self.events_seen = int(events_seen) + self.saw_message_delta = bool(saw_message_delta) + self.block_types = list(block_types) + self.elapsed_s = float(elapsed_s) + super().__init__( + "truncated stream: ended without message_stop " + f"(last_event={last_event or 'none'}, events_seen={self.events_seen}, " + f"message_delta={'yes' if self.saw_message_delta else 'no'}, " + f"blocks=[{','.join(self.block_types)}], " + f"elapsed={self.elapsed_s:.1f}s)", + ) + + +class LLMOpenAITruncatedStream(LLMError): + """An OpenAI-compatible stream closed before a completion signal. + + Chat Completions completes with ``[DONE]`` or a ``finish_reason``; + Responses with ``response.completed`` / ``response.incomplete``. Routed + like :class:`LLMTruncatedStream`. + """ + + def __init__( + self, *, protocol: str, last_event: str, events_seen: int, + expected: str, elapsed_s: float, + ) -> None: + self.protocol = protocol + self.last_event = last_event + self.events_seen = int(events_seen) + self.expected = expected + self.elapsed_s = float(elapsed_s) + super().__init__( + f"truncated stream: {protocol} ended without {expected} " + f"(last_event={last_event or 'none'}, events_seen={self.events_seen}, " + f"elapsed={self.elapsed_s:.1f}s)", + ) + + class LLMStreamStalled(LLMError, TimeoutError): """A streaming LLM call went silent mid-flight. @@ -191,8 +262,10 @@ def __init__( "LLMDeadlineExceeded", "LLMEmptyCompletion", "LLMError", + "LLMOpenAITruncatedStream", "LLMReasoningRunaway", "LLMStreamStalled", + "LLMTruncatedStream", "PermissionDenied", "ServiceNotRegistered", "TaskNotFoundError", diff --git a/agent_core/providers/_stream_activity.py b/agent_core/providers/_stream_activity.py index 17da4ba..3476972 100644 --- a/agent_core/providers/_stream_activity.py +++ b/agent_core/providers/_stream_activity.py @@ -5,6 +5,8 @@ import asyncio import contextlib import functools +import os +import re from collections.abc import AsyncGenerator, AsyncIterator, Callable from typing import Any @@ -13,6 +15,8 @@ from agent_core.runtime.async_utils import await_bounded _CLEANUP_TIMEOUT_S = 5.0 +_SSE_LINE_BREAK = re.compile(rb"[\r\n]") +STREAM_TERMINATOR_ENV = "AGENT_CORE_STREAM_REQUIRE_TERMINATOR" class StreamActivity: @@ -20,10 +24,113 @@ class StreamActivity: def __init__(self) -> None: self.output_reported = False + # The OpenAI SDK consumes the Chat Completions [DONE] sentinel without + # yielding it. Keep this wire-level signal for the adapter's end check; + # ``done_observable`` is False when no observer could be installed. + self.saw_done = False + self.done_observable = False + self._line = b"" + self._line_long = False def mark_output(self) -> None: self.output_reported = True + def observe_bytes(self, chunk: bytes) -> None: + """Scan DECODED SSE bytes for the ``data: [DONE]`` line.""" + if self.saw_done: + return + # SSE ends a line on CRLF, LF or a lone CR. Splitting on both bytes + # turns CRLF into an extra empty line, which is harmless here. + parts = _SSE_LINE_BREAK.split(chunk) + for part in parts[:-1]: + self._append_line(part) + self._check_line() + self._line = b"" + self._line_long = False + self._append_line(parts[-1]) + + def finish_bytes(self) -> None: + """The body ended: a final line without its line break still counts.""" + if not self.saw_done: + self._check_line() + self._line = b"" + self._line_long = False + + def _check_line(self) -> None: + # Same rule as the SDK: field ``data``, one optional leading space, + # value starting with ``[DONE]``. + if self._line_long or not self._line.startswith(b"data:"): + return + value = self._line[5:] + if value.startswith(b" "): + value = value[1:] + if value.startswith(b"[DONE]"): + self.saw_done = True + + def _append_line(self, part: bytes) -> None: + # Only the short sentinel line matters; never retain an unbounded SSE + # JSON line containing model output. + if not self._line_long: + if len(self._line) + len(part) <= 32: + self._line += part + else: + self._line = b"" + self._line_long = True + + +class _DoneSentinelDecoder: + """Feed the SDK's SSE decoder through :meth:`StreamActivity.observe_bytes`. + + The decoder receives ``response.aiter_bytes()`` — already decompressed — + whereas ``response.stream`` carries the raw body, where a gzip/deflate + encoded SSE stream hides the sentinel entirely. + """ + + def __init__(self, inner: Any, activity: StreamActivity) -> None: + self._inner = inner + self._activity = activity + + def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[Any]: + activity = self._activity + + async def observed() -> AsyncIterator[bytes]: + async for chunk in iterator: + activity.observe_bytes(chunk) + yield chunk + activity.finish_bytes() + + return self._inner.aiter_bytes(observed()) + + def __getattr__(self, name: str) -> Any: + return getattr(self._inner, name) + + +def watch_done_sentinel(stream: Any, activity: StreamActivity) -> None: + """Record whether an OpenAI SDK stream delivered ``data: [DONE]``. + + Must run before the stream is iterated: the SDK reads ``_decoder`` when its + event iterator starts. An object without one (a custom/test iterator) + leaves ``done_observable`` False, and the adapter must not demand it. + """ + decoder = getattr(stream, "_decoder", None) + if decoder is None or not callable(getattr(decoder, "aiter_bytes", None)): + return + try: + stream._decoder = _DoneSentinelDecoder(decoder, activity) + except (AttributeError, TypeError): + return + activity.done_observable = True + + +def stream_terminator_required() -> bool: + """Whether a stream that ends without its protocol terminator raises. + + ``AGENT_CORE_STREAM_REQUIRE_TERMINATOR=0`` is the escape hatch for a + gateway that never sends one: the adapters then log the missing + terminator and accept the turn, which is the pre-0.14.3 behaviour. + """ + return os.getenv(STREAM_TERMINATOR_ENV, "1").strip() != "0" + class _ActivityByteStream(httpx.AsyncByteStream): def __init__(self, inner: Any, on_bytes: Callable[[], None]) -> None: diff --git a/agent_core/providers/anthropic.py b/agent_core/providers/anthropic.py index ae4e18d..839f77a 100644 --- a/agent_core/providers/anthropic.py +++ b/agent_core/providers/anthropic.py @@ -25,13 +25,20 @@ import json import logging import os +import time from collections.abc import AsyncIterator from typing import Any +from agent_core.errors import LLMTruncatedStream from agent_core.llm import LLMClient, LLMResponse, StreamDelta from agent_core.messages import Message, ToolCall, text_of from agent_core.model_capabilities import ModelCapabilities, resolve_model_capabilities -from agent_core.providers._stream_activity import StreamActivity, stream_events_with_activity +from agent_core.providers._stream_activity import ( + STREAM_TERMINATOR_ENV, + StreamActivity, + stream_events_with_activity, + stream_terminator_required, +) from agent_core.providers.finish_reason import normalize_finish_reason logger = logging.getLogger(__name__) @@ -253,6 +260,13 @@ async def stream( # string, so a streamed thinking turn used to yield reasoning that # ``thinking_format="content_block"`` could not replay. blocks: dict[int, dict[str, Any]] = {} + # End record: the protocol always closes with ``message_delta`` then + # ``message_stop``; a stream that ends before the latter was cut off. + events_seen = 0 + last_event = "" + saw_message_delta = False + saw_message_stop = False + started = time.monotonic() stream, reset = await self._create_message(kwargs) activity = StreamActivity() async with contextlib.aclosing(stream_events_with_activity(stream, activity)) as events: @@ -263,7 +277,11 @@ async def stream( yield StreamDelta(transport_activity=True) continue etype = getattr(event, "type", "") - if etype == "message_start": + events_seen += 1 + last_event = etype + if etype == "message_stop": + saw_message_stop = True + elif etype == "message_start": msg = getattr(event, "message", None) model = getattr(msg, "model", "") or model u = getattr(msg, "usage", None) @@ -354,6 +372,7 @@ async def stream( "arguments": getattr(d, "partial_json", "") or "", }]) elif etype == "message_delta": + saw_message_delta = True d = getattr(event, "delta", None) stop_reason = getattr(d, "stop_reason", "") or stop_reason # A classifier refusal ends the stream here, with the detail on @@ -375,6 +394,24 @@ async def stream( rt = _anthropic_reasoning_tokens(u) if rt is not None: reasoning_tokens = rt + if not saw_message_stop: + # Raised, not returned: whatever arrived (often a complete thinking + # block) is shaped like a deliberate stop, and ``no_tool`` would + # end the run on it. Nothing has executed yet, so a resample is + # safe; the error text carries the end record for the retry log. + error = LLMTruncatedStream( + last_event=last_event, + events_seen=events_seen, + saw_message_delta=saw_message_delta, + block_types=[ + str(b.get("type", "")) for b in _ordered_blocks(blocks) + ], + elapsed_s=time.monotonic() - started, + ) + if stream_terminator_required(): + raise error + logger.warning("Accepting stream without terminator (%s=0): %s", + STREAM_TERMINATOR_ENV, error) # Terminal delta: fold the accumulated usage/finish/model onto the # assembled ``LLMResponse`` (mirrors OpenAI's empty-choices chunk). # ``reasoning_blocks`` is sent ONLY for a thinking turn — for a plain diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index aef522e..8308150 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -68,6 +68,7 @@ # side has to import the other's package for them — see that module. from agent_core.retry_policy import LEGACY_RETRYABLE_KEYWORDS, legacy_retryable from agent_core.runtime.async_utils import closing_stream +from agent_core.runtime.retriable import is_truncated_stream logger = logging.getLogger(__name__) @@ -210,6 +211,8 @@ class CooldownFallbackLLM: consumer, because a retry would emit those deltas twice. Set ``replay_partial_stream=True`` to restore the historical duplicating behavior; prefer :class:`LLMFallbackChain` for rewind-safe semantics. + A truncated stream that already yielded deltas propagates its error so + the caller can start a fresh attempt instead of combining model outputs. Event hook contract ------------------- @@ -532,6 +535,11 @@ async def stream( ) if not self._retryable(error): break + if yielded and is_truncated_stream(error): + # The consumer already has part of this turn. Appending a + # fallback turn would turn two responses into one apparent + # success; let the caller discard and restart the attempt. + raise if yielded and not self._replay_partial_stream: # Deltas already reached the consumer; retrying the primary # would duplicate them. Degrade to the fallback leg instead. diff --git a/agent_core/providers/openai_chat.py b/agent_core/providers/openai_chat.py index 507fc34..fe49ba6 100644 --- a/agent_core/providers/openai_chat.py +++ b/agent_core/providers/openai_chat.py @@ -15,15 +15,23 @@ # pyright: basic, reportPrivateImportUsage=false import contextlib import logging +import time from collections.abc import AsyncIterator, Callable from typing import Any from openai import AsyncOpenAI, BadRequestError +from agent_core.errors import LLMOpenAITruncatedStream from agent_core.llm import LLMClient, LLMResponse, StreamDelta from agent_core.messages import Message, ToolCall, for_wire from agent_core.providers._api_key import resolve_openai_api_key -from agent_core.providers._stream_activity import StreamActivity, stream_events_with_activity +from agent_core.providers._stream_activity import ( + STREAM_TERMINATOR_ENV, + StreamActivity, + stream_events_with_activity, + stream_terminator_required, + watch_done_sentinel, +) from agent_core.runtime.llm_request_overrides import ( current_thinking_retry_override, ) @@ -392,11 +400,16 @@ async def stream( stream = await self._open_stream(kwargs) activity = StreamActivity() + watch_done_sentinel(stream, activity) + started = time.monotonic() + chunks_seen = 0 + last_finish_reason = "" async with contextlib.aclosing(stream_events_with_activity(stream, activity)) as events: async for chunk in events: if chunk is None: yield StreamDelta(transport_activity=True) continue + chunks_seen += 1 chunk_usage = _usage_dict(getattr(chunk, "usage", None)) chunk_model = getattr(chunk, "model", "") or "" if not chunk.choices: @@ -409,15 +422,32 @@ async def stream( continue choice = chunk.choices[0] delta = choice.delta + finish_reason = getattr(choice, "finish_reason", None) or "" + if finish_reason: + last_finish_reason = finish_reason activity.mark_output() yield StreamDelta( content=getattr(delta, "content", None) or "", reasoning_content=_reasoning_text(delta), tool_call_deltas=_tool_call_deltas(getattr(delta, "tool_calls", None)), - finish_reason=getattr(choice, "finish_reason", None) or "", + finish_reason=finish_reason, model=chunk_model, usage=chunk_usage, ) + # Some compatible gateways omit finish_reason, while the SDK consumes + # [DONE] without yielding it. Either signal proves a completed turn. + # When [DONE] cannot be observed (no SDK decoder to watch), a missing + # finish_reason proves nothing and the turn is accepted as before. + if activity.done_observable and not (activity.saw_done or last_finish_reason): + error = LLMOpenAITruncatedStream( + protocol="chat_completions", last_event="chunk" if chunks_seen else "", + events_seen=chunks_seen, expected="[DONE] or finish_reason", + elapsed_s=time.monotonic() - started, + ) + if stream_terminator_required(): + raise error + logger.warning("Accepting stream without terminator (%s=0): %s", + STREAM_TERMINATOR_ENV, error) async def _open_stream(self, kwargs: dict[str, Any]) -> Any: """Open the stream, retrying once past a rejected ``reasoning_effort``.""" diff --git a/agent_core/providers/openai_responses.py b/agent_core/providers/openai_responses.py index edec592..a76b2c9 100644 --- a/agent_core/providers/openai_responses.py +++ b/agent_core/providers/openai_responses.py @@ -31,16 +31,22 @@ # pyright: basic, reportPrivateImportUsage=false import contextlib import logging +import time from collections.abc import AsyncIterator from typing import Any from openai import AsyncOpenAI -from agent_core.errors import LLMError +from agent_core.errors import LLMError, LLMOpenAITruncatedStream from agent_core.llm import LLMClient, LLMResponse, StreamDelta from agent_core.messages import Message, ToolCall, text_of from agent_core.providers._api_key import resolve_openai_api_key -from agent_core.providers._stream_activity import StreamActivity, stream_events_with_activity +from agent_core.providers._stream_activity import ( + STREAM_TERMINATOR_ENV, + StreamActivity, + stream_events_with_activity, + stream_terminator_required, +) from agent_core.providers.finish_reason import ( normalize_finish_reason, responses_finish_reason, @@ -190,12 +196,18 @@ async def stream( kwargs["stream"] = True stream = await self._client.responses.create(**kwargs) activity = StreamActivity() + started = time.monotonic() + events_seen = 0 + last_event = "" + saw_terminal = False async with contextlib.aclosing(stream_events_with_activity(stream, activity)) as events: async for event in events: if event is None: yield StreamDelta(transport_activity=True) continue etype = getattr(event, "type", "") + events_seen += 1 + last_event = etype if etype == "response.output_text.delta": activity.mark_output() yield StreamDelta(content=getattr(event, "delta", "") or "") @@ -205,11 +217,20 @@ async def stream( ): activity.mark_output() yield StreamDelta(reasoning_content=getattr(event, "delta", "") or "") + elif etype == "error": + # A stream-level failure carries code/message at the top + # level, not under ``response.error``. Surface it as-is so + # the retry classifier sees the real cause. + raise _ResponsesError( + str(getattr(event, "code", "") or ""), + getattr(event, "message", "") or "Responses stream error", + ) elif etype == "response.failed": raise _response_failure( getattr(event, "response", None), fallback="Responses request failed", ) elif etype in ("response.completed", "response.incomplete"): + saw_terminal = True resp = getattr(event, "response", None) usage = _responses_usage_dict(getattr(resp, "usage", None)) reason = normalize_finish_reason( @@ -221,6 +242,17 @@ async def stream( model=getattr(resp, "model", "") or "", finish_reason=reason or "stop", ) + if not saw_terminal: + error = LLMOpenAITruncatedStream( + protocol="responses", last_event=last_event, + events_seen=events_seen, + expected="response.completed or response.incomplete", + elapsed_s=time.monotonic() - started, + ) + if stream_terminator_required(): + raise error + logger.warning("Accepting stream without terminator (%s=0): %s", + STREAM_TERMINATOR_ENV, error) # ── Conversion helpers (pure — unit-tested) ──────────────────────────────── diff --git a/agent_core/retry_policy.py b/agent_core/retry_policy.py index 17e71c6..1a76fb9 100644 --- a/agent_core/retry_policy.py +++ b/agent_core/retry_policy.py @@ -17,7 +17,7 @@ "timeout", "timed out", "429", "500", "502", "503", "504", "529", "overloaded", "rate limit", "rate_limit", "server error", "connection reset", "connection error", "econnreset", "gateway timeout", - "model_dump", "model_not_found", + "model_dump", "model_not_found", "truncated stream", }) diff --git a/agent_core/runtime/loop/_call.py b/agent_core/runtime/loop/_call.py index 389ec6c..8bac92f 100644 --- a/agent_core/runtime/loop/_call.py +++ b/agent_core/runtime/loop/_call.py @@ -298,6 +298,7 @@ async def call_llm( is_empty_completion, is_overloaded_error, is_retriable_with_fallback, + is_truncated_stream, ) def _transient_backoff(attempt: int) -> float: @@ -1095,8 +1096,8 @@ async def _finish_attempt( # Skip the rest of the retry budget and surface so an outer # chain wrapper can advance the leg right now. if is_retriable_with_fallback(exc): - # Overload (503) and empty completions frequently clear on a - # same-key resample (temperature>0 re-rolls the sampler). When + # Overload (503), empty completions and truncated streams + # frequently clear on a same-key resample (temperature>0 re-rolls the sampler). When # NO outer chain is active to advance a leg # (``chain_fallback_active()`` is False), # short-circuiting these would trade a recoverable blip for a @@ -1107,6 +1108,7 @@ async def _finish_attempt( # credit / safety) where retrying the same key cannot help. resample_may_recover = ( is_overloaded_error(exc) or is_empty_completion(exc) + or is_truncated_stream(exc) ) if _chain_fallback_active() or not resample_may_recover: logger.error( diff --git a/agent_core/runtime/retriable.py b/agent_core/runtime/retriable.py index 7c7a361..01f0140 100644 --- a/agent_core/runtime/retriable.py +++ b/agent_core/runtime/retriable.py @@ -151,6 +151,15 @@ re.compile(r"upstream[\s_]*(?:timeout|error)", re.IGNORECASE), ) +# ``LLMTruncatedStream`` / ``LLMOpenAITruncatedStream``: the stream closed +# cleanly before its protocol terminator. Routed like an empty completion: a +# same-key resample usually clears it, and when it does not (or a chain is +# active) a different provider is the guaranteed recovery. +_TRUNCATED_STREAM_PATTERNS = ( + re.compile(r"\bLLM(?:OpenAI)?TruncatedStream\b"), + re.compile(r"truncated[\s_]*stream", re.IGNORECASE), +) + # Runtime stream watchdog. Matched by type name / message text instead of # importing ``LLMStreamStalled`` from core to keep infra free of core imports. _STREAM_STALL_PATTERNS = ( @@ -351,6 +360,8 @@ def is_transient_network(err: BaseException) -> bool: return False if is_stream_stall(err): return False + if is_truncated_stream(err): + return False if is_overloaded_error(err): return False # model_unavailable is also a 5xx (typically 503 from distributor @@ -461,6 +472,17 @@ def is_empty_completion(err: BaseException) -> bool: return any(p.search(blob) for p in _EMPTY_COMPLETION_PATTERNS) +def is_truncated_stream(err: BaseException) -> bool: + """True if a stream closed cleanly without its protocol terminator. + + What arrived is well-formed but partial, so it must never be read as a + deliberate stop. Routed like :func:`is_empty_completion`: same-key + resample first, chain advance when that is the better recovery. + """ + blob = _stringify(err) + return any(p.search(blob) for p in _TRUNCATED_STREAM_PATTERNS) + + def is_retriable_with_fallback(err: BaseException) -> bool: """The chain-escalation trigger. @@ -471,8 +493,8 @@ def is_retriable_with_fallback(err: BaseException) -> bool: is deterministic on (current provider, current input) — only switching providers / keys can change the outcome. - ``empty_completion`` also routes here: it isn't deterministic on the - input (a temp>0 resample may recover), but the caller's same-key + ``empty_completion`` and ``truncated_stream`` also route here: they + aren't deterministic on the input (a temp>0 resample may recover), but the caller's same-key retry budget runs first, and advancing the chain afterwards is the guaranteed recovery — so it belongs to the same predicate. """ @@ -483,6 +505,7 @@ def is_retriable_with_fallback(err: BaseException) -> bool: or is_model_unavailable(err) or is_auth_failure(err) or is_empty_completion(err) + or is_truncated_stream(err) or is_stream_stall(err) ) @@ -493,7 +516,8 @@ def classify_error(err: BaseException) -> str: Precedence (top wins): runtime deadline → ``context_length`` → ``safety_filter`` → ``model_unavailable`` → - ``auth_failure`` → ``empty_completion`` → ``overloaded`` → + ``auth_failure`` → ``empty_completion`` → ``truncated_stream`` → + ``overloaded`` → ``credit_exhausted`` → ``stream_stall`` → ``rate_limited`` → ``transient_network`` → ``other``. @@ -523,6 +547,8 @@ def classify_error(err: BaseException) -> str: return "auth_failure" if is_empty_completion(err): return "empty_completion" + if is_truncated_stream(err): + return "truncated_stream" if is_overloaded_error(err): return "overloaded" if is_credit_exhausted(err): @@ -550,4 +576,5 @@ def classify_error(err: BaseException) -> str: "is_safety_filter", "is_stream_stall", "is_transient_network", + "is_truncated_stream", ] diff --git a/changes/71.fix.md b/changes/71.fix.md new file mode 100644 index 0000000..31166ea --- /dev/null +++ b/changes/71.fix.md @@ -0,0 +1 @@ +Retry Anthropic and OpenAI-compatible streams that close before a completion signal, preventing partial responses from silently ending an agent run. Truncated streams are classified `truncated_stream`: resampled on the same key, or advanced to the next provider when a fallback chain is active. The Chat Completions `[DONE]` check reads decoded bytes, so gzip/deflate-encoded event streams are handled. Responses `error` events now surface their code and message. `AGENT_CORE_STREAM_REQUIRE_TERMINATOR=0` accepts unterminated streams with a warning, for gateways that never send a terminator. diff --git a/tests/test_anthropic_truncated_stream.py b/tests/test_anthropic_truncated_stream.py new file mode 100644 index 0000000..a24e9e5 --- /dev/null +++ b/tests/test_anthropic_truncated_stream.py @@ -0,0 +1,182 @@ +"""An Anthropic stream that ends before ``message_stop`` is a dropped connection. + +Measured on ApodexHarness's 2026-10-07 GDPval batch over llm-hub: 10 of 15 +finished trials ended on a turn shaped exactly like the fixture below -- +``message_start`` usage, a thinking block, then nothing: no text, no tool call, +no ``message_delta``. Several unrelated in-flight streams were cut in the same +second, three times. The SDK raised nothing (no SSE ``error`` event, no broken +chunked body), the assembled response carried usage so the empty-completion +guard let it through, and ``no_tool`` ended each run with its work discarded. +""" +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from agent_core.errors import LLMTruncatedStream +from agent_core.loop_types import LoopConfig, LoopPolicy +from agent_core.messages import user_msg +from agent_core.providers import anthropic as ac +from agent_core.providers._stream_activity import STREAM_TERMINATOR_ENV +from agent_core.retry_policy import legacy_retryable +from agent_core.runtime.loop.agent_loop import run_agent_loop +from agent_core.runtime.loop.model_profile import ModelProfile +from agent_core.runtime.retriable import ( + classify_error, + is_empty_completion, + is_retriable_with_fallback, + is_transient_network, + is_truncated_stream, +) + + +def _message_start() -> SimpleNamespace: + return SimpleNamespace(type="message_start", message=SimpleNamespace( + model="claude-x", + usage=SimpleNamespace(input_tokens=158, cache_read_input_tokens=9326), + )) + + +def _thinking(idx: int = 0) -> list[SimpleNamespace]: + return [ + SimpleNamespace(type="content_block_start", index=idx, + content_block=SimpleNamespace(type="thinking")), + SimpleNamespace(type="content_block_delta", index=idx, delta=SimpleNamespace( + type="thinking_delta", thinking="I need to write the STEP file next.")), + SimpleNamespace(type="content_block_delta", index=idx, delta=SimpleNamespace( + type="signature_delta", signature="sig")), + SimpleNamespace(type="content_block_stop", index=idx), + ] + + +def _truncated() -> list[SimpleNamespace]: + """The measured shape: usage and thinking, then the body just ends.""" + return [_message_start(), *_thinking()] + + +def _tool_call_turn() -> list[SimpleNamespace]: + return [ + _message_start(), + *_thinking(), + SimpleNamespace(type="content_block_start", index=1, content_block=SimpleNamespace( + type="tool_use", id="toolu_1", name="bash", input={})), + SimpleNamespace(type="content_block_delta", index=1, delta=SimpleNamespace( + type="input_json_delta", partial_json='{"command": "ls"}')), + SimpleNamespace(type="content_block_stop", index=1), + SimpleNamespace(type="message_delta", delta=SimpleNamespace(stop_reason="tool_use"), + usage=SimpleNamespace(output_tokens=40)), + SimpleNamespace(type="message_stop"), + ] + + +def _final_text_turn() -> list[SimpleNamespace]: + return [ + _message_start(), + SimpleNamespace(type="content_block_start", index=0, + content_block=SimpleNamespace(type="text")), + SimpleNamespace(type="content_block_delta", index=0, + delta=SimpleNamespace(type="text_delta", text="done")), + SimpleNamespace(type="content_block_stop", index=0), + SimpleNamespace(type="message_delta", delta=SimpleNamespace(stop_reason="end_turn"), + usage=SimpleNamespace(output_tokens=2)), + SimpleNamespace(type="message_stop"), + ] + + +def _client(*streams: list[SimpleNamespace]) -> ac.AnthropicClient: + pending = list(streams) + + async def create(**_kwargs: Any) -> Any: + events = pending.pop(0) + + async def gen() -> Any: + for event in events: + yield event + + return gen() + + c = ac.AnthropicClient("claude-x", api_key="x") + c._client = SimpleNamespace(messages=SimpleNamespace(create=AsyncMock(side_effect=create))) + return c + + +@pytest.mark.asyncio +async def test_stream_without_message_stop_raises_with_its_end_record() -> None: + c = _client(_truncated()) + + with pytest.raises(LLMTruncatedStream) as info: + _ = [d async for d in c.stream([user_msg("go")])] + + err = info.value + assert err.last_event == "content_block_stop" + assert err.events_seen == 5 + assert err.saw_message_delta is False + assert err.block_types == ["thinking"] + assert "ended without message_stop" in str(err) + + +@pytest.mark.asyncio +async def test_message_delta_alone_is_still_truncated() -> None: + """``message_stop`` is the terminator; a stop_reason without it was cut too.""" + events = [*_tool_call_turn()[:-1]] # drop only message_stop + c = _client(events) + + with pytest.raises(LLMTruncatedStream) as info: + _ = [d async for d in c.stream([user_msg("go")])] + + assert info.value.saw_message_delta is True + assert info.value.last_event == "message_delta" + + +@pytest.mark.asyncio +async def test_complete_stream_is_untouched() -> None: + deltas = [d async for d in _client(_tool_call_turn()).stream([user_msg("go")])] + assert deltas[-1].stop_reason == "tool_use" + + +def test_classified_as_truncated_stream() -> None: + """Resample on the same key; an active chain advances, like empty completion.""" + err = LLMTruncatedStream(last_event="content_block_stop", events_seen=5, + saw_message_delta=False, block_types=["thinking"], + elapsed_s=776.0) + assert is_truncated_stream(err) + assert classify_error(err) == "truncated_stream" + assert is_retriable_with_fallback(err) + assert not is_transient_network(err) + assert not is_empty_completion(err) + assert legacy_retryable(err) + + +@pytest.mark.asyncio +async def test_escape_hatch_accepts_the_partial_turn(monkeypatch, caplog) -> None: + monkeypatch.setenv(STREAM_TERMINATOR_ENV, "0") + with caplog.at_level("WARNING"): + deltas = [d async for d in _client(_truncated()).stream([user_msg("go")])] + assert deltas[-1].usage + assert "ended without message_stop" in caplog.text + + +@pytest.mark.asyncio +async def test_loop_resamples_a_truncated_turn_instead_of_ending_the_run() -> None: + llm = _client(_truncated(), _tool_call_turn(), _final_text_turn()) + tool = MagicMock() + tool.name = "bash" + tool.ainvoke = AsyncMock(return_value="ok") + + result = await run_agent_loop( + system_prompt="system", user_message="start", llm=llm, tools=[tool], + config=LoopConfig( + max_turns=5, max_llm_retries=3, retry_wait_fixed=0, + stream_llm_tokens=True, + loop_policy=LoopPolicy(no_tool_behavior="stop"), + ), + model_profile=ModelProfile(model_id="claude-x", provider="p", protocol="anthropic"), + ) + + assert llm._client.messages.create.await_count == 3 + tool.ainvoke.assert_awaited_once() + assert result.stopped_by == "no_tool" + assert result.final_content == "done" diff --git a/tests/test_cooldown_fallback.py b/tests/test_cooldown_fallback.py index 8f8164c..fc10d59 100644 --- a/tests/test_cooldown_fallback.py +++ b/tests/test_cooldown_fallback.py @@ -4,9 +4,11 @@ import pytest +from agent_core.errors import LLMOpenAITruncatedStream from agent_core.llm import LLMResponse, StreamDelta from agent_core.messages import Message, user_msg from agent_core.providers.fallback import CooldownFallbackLLM, legacy_retryable +from agent_core.runtime.loop.llm_client import call_llm class ScriptedLLM: @@ -166,6 +168,49 @@ async def stream( yield StreamDelta(content="FALLBACK") +class TruncatingStream: + model = "primary" + + def __init__(self) -> None: + self.starts = 0 + + async def stream(self, _messages: list[Message], **_kwargs: object) -> AsyncIterator[StreamDelta]: + self.starts += 1 + if self.starts == 1: + yield StreamDelta(content="partial ") + raise LLMOpenAITruncatedStream( + protocol="chat_completions", last_event="chunk", events_seen=1, + expected="[DONE] or finish_reason", elapsed_s=1.0, + ) + yield StreamDelta(content="complete", finish_reason="stop") + + +@pytest.mark.asyncio +async def test_truncated_partial_stream_does_not_append_fallback_turn() -> None: + primary = TruncatingStream() + llm = CooldownFallbackLLM(primary, OkStream(), max_retries=1, clock=lambda: 0.0) + + with pytest.raises(LLMOpenAITruncatedStream): + _ = [delta async for delta in llm.stream([user_msg("x")])] + + assert primary.starts == 1 + assert llm._cooldown_until == 0.0 + + async def ignore(*_args: object, **_kwargs: object) -> None: + pass + + retry_primary = TruncatingStream() + retry_llm = CooldownFallbackLLM( + retry_primary, OkStream(), max_retries=1, clock=lambda: 0.0, + ) + response = await call_llm( + retry_llm, [user_msg("x")], timeout=5, max_retries=2, turn=1, + on_delta=ignore, retry_wait_fixed=0, + ) + assert response.content == "complete" + assert retry_primary.starts == 2 + + @pytest.mark.asyncio async def test_stream_does_not_replay_already_yielded_deltas() -> None: primary = PartialStream() diff --git a/tests/test_openai_stream_activity.py b/tests/test_openai_stream_activity.py index ddebdcc..b1c8823 100644 --- a/tests/test_openai_stream_activity.py +++ b/tests/test_openai_stream_activity.py @@ -11,12 +11,13 @@ from agent_core.components.middleware.llm.base import LLMMiddleware, LLMMiddlewareChain from agent_core.components.middleware.llm.proxy import LLMProxy -from agent_core.errors import LLMReasoningRunaway, LLMStreamStalled +from agent_core.errors import LLMOpenAITruncatedStream, LLMReasoningRunaway, LLMStreamStalled from agent_core.messages import user_msg from agent_core.providers.fallback import FallbackEntry, LLMFallbackChain from agent_core.providers.openai_chat import OpenAIClient from agent_core.providers.openai_responses import OpenAIResponsesClient from agent_core.runtime.loop._streaming import _stream_llm_response +from agent_core.runtime.retriable import classify_error, is_retriable_with_fallback if issubclass(DefaultAsyncHttpxClient, httpx.AsyncClient): sdk_httpx = httpx @@ -97,7 +98,8 @@ async def _read(self): yield _sse({"type": "response.reasoning_summary_text.delta", "item_id": "r1", "output_index": 0, "summary_index": 0, "delta": "thinking", "sequence_number": 0}, "response.reasoning_summary_text.delta") - beats = {"dense": 0, "long": 100}.get(self.mode, 30) + beats = {"dense": 0, "long": 100, "truncated": 0, + "done_only": 0, "finish_only": 0}.get(self.mode, 30) for _ in range(beats): await asyncio.sleep(0.01) self.pings += 1 @@ -116,7 +118,25 @@ async def _read(self): "response.in_progress") else: yield b": keep-alive\n\n" - if self.mode == "error": + if self.mode == "truncated": + protocol, model = self.profile + if protocol == "chat": + yield _sse(_chat_chunk(model, {"reasoning_content": "partial thought"})) + else: + yield _sse({"type": "response.reasoning_summary_text.delta", + "delta": "partial thought", "sequence_number": 1}, + "response.reasoning_summary_text.delta") + elif self.mode == "done_only": + protocol, model = self.profile + assert protocol == "chat" + yield _sse(_chat_chunk(model, {"content": "done"})) + yield b"data: [DO" + yield b"NE]\n\n" + elif self.mode == "finish_only": + protocol, model = self.profile + assert protocol == "chat" + yield _sse(_chat_chunk(model, {"content": "done"}, "stop")) + elif self.mode == "error": yield _sse({"error": {"message": "upstream unavailable", "type": "server_error"}}) else: for chunk in _ending(*self.profile, tool=self.tool): @@ -183,6 +203,44 @@ async def on_delta(text, accumulated, index, thinking, **kwargs): assert requests[0]["stream"] is True and requests[0]["model"] == profile[1] +async def test_cleanly_closed_short_stream_is_rejected(profile): + async with _client(profile, ("truncated",)) as (client, bodies, _): + with pytest.raises(LLMOpenAITruncatedStream) as info: + await _stream_llm_response(client, [user_msg("hi")], 2, _ignore) + assert bodies[0].closed + assert info.value.events_seen == 1 + assert info.value.protocol == ("chat_completions" if profile[0] == "chat" else "responses") + assert classify_error(info.value) == "truncated_stream" + assert is_retriable_with_fallback(info.value) + + +@pytest.mark.parametrize("mode", ["done_only", "finish_only"]) +async def test_compatible_chat_completion_accepts_either_end_signal(mode): + async with _client(("chat", "apodex-1.1-mini"), (mode,)) as (client, _, _): + response = await _stream_llm_response(client, [user_msg("hi")], 2, _ignore) + assert response.content == "done" + + +async def test_truncated_chat_turn_is_retried_before_no_tool_stop(): + from agent_core.loop_types import LoopConfig, LoopPolicy + from agent_core.runtime.loop.agent_loop import run_agent_loop + + async with _client(("chat", "apodex-1.1-mini"), ("truncated", "healthy")) as ( + client, bodies, requests, + ): + result = await run_agent_loop( + system_prompt="s", user_message="hi", llm=client, tools=[], + config=LoopConfig(max_turns=2, max_llm_retries=2, + retry_wait_fixed=0, llm_timeout=2, + stream_llm_tokens=True, + loop_policy=LoopPolicy(no_tool_behavior="stop")), + ) + assert len(requests) == 2 + assert all(body.closed for body in bodies) + assert result.stopped_by == "no_tool" + assert result.final_content == "done" + + async def test_silent_socket_still_stalls_and_closes(profile): async with _client(profile, ("silent",)) as (client, bodies, _): with pytest.raises(LLMStreamStalled): diff --git a/tests/test_provider_native_clients.py b/tests/test_provider_native_clients.py index 884d1b3..5838cab 100644 --- a/tests/test_provider_native_clients.py +++ b/tests/test_provider_native_clients.py @@ -414,7 +414,7 @@ async def test_openai_stream_mirrors_session_into_extra_query(): async def fake_stream(): yield SimpleNamespace(choices=[SimpleNamespace( delta=SimpleNamespace(content="ok", reasoning_content=None, - tool_calls=None))]) + tool_calls=None), finish_reason="stop")]) create = AsyncMock(return_value=fake_stream()) c._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace( @@ -461,7 +461,8 @@ async def fake_stream(): content="llo", reasoning_content="r", tool_calls=[SimpleNamespace( index=0, id="c1", - function=SimpleNamespace(name="f", arguments='{}'))]))]) + function=SimpleNamespace(name="f", arguments='{}'))]), + finish_reason="tool_calls")]) c._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace( create=AsyncMock(return_value=fake_stream()), @@ -1139,6 +1140,7 @@ async def fake_events(): yield SimpleNamespace(type="message_delta", delta=SimpleNamespace(stop_reason="end_turn"), usage=SimpleNamespace(output_tokens=2)) + yield SimpleNamespace(type="message_stop") c = ac.AnthropicClient("claude-x", api_key="x") c._client = SimpleNamespace( @@ -1480,6 +1482,7 @@ async def test_anthropic_streamed_refusal_reports_the_same_detail(): ), usage=SimpleNamespace(output_tokens=0), ), + SimpleNamespace(type="message_stop"), ] class _Stream: diff --git a/tests/test_runaway_retry_transient.py b/tests/test_runaway_retry_transient.py index 6266ece..2140347 100644 --- a/tests/test_runaway_retry_transient.py +++ b/tests/test_runaway_retry_transient.py @@ -90,6 +90,7 @@ async def events() -> Any: delta=SimpleNamespace(stop_reason="end_turn" if done else "max_tokens"), usage=SimpleNamespace(output_tokens=2 if done else 2048), ) + yield SimpleNamespace(type="message_stop") return events() diff --git a/tests/test_stream_terminator_edges.py b/tests/test_stream_terminator_edges.py new file mode 100644 index 0000000..c9d43b7 --- /dev/null +++ b/tests/test_stream_terminator_edges.py @@ -0,0 +1,269 @@ +"""Edge cases of the stream-terminator check across the real OpenAI SDK. + +The Chat Completions ``[DONE]`` sentinel is consumed by the SDK, so the adapter +watches the bytes the SDK's SSE decoder reads. Those are DECODED bytes: a +gateway that gzip/deflate-encodes its event stream must not hide the sentinel, +and every SSE line ending the SDK accepts must count. +""" +from __future__ import annotations + +import gzip +import zlib +from types import SimpleNamespace + +import pytest +from openai import AsyncOpenAI + +from agent_core.errors import LLMOpenAITruncatedStream, LLMTruncatedStream +from agent_core.llm import StreamDelta +from agent_core.messages import user_msg +from agent_core.providers._stream_activity import ( + STREAM_TERMINATOR_ENV, + StreamActivity, +) +from agent_core.providers.openai_chat import OpenAIClient +from agent_core.providers.openai_responses import OpenAIResponsesClient +from agent_core.runtime.loop._streaming import _stream_llm_response +from agent_core.runtime.loop.llm_client import LLMCallExhausted, call_llm +from agent_core.runtime.retriable import classify_error, is_transient_network +from tests.test_openai_stream_activity import _chat_chunk, _sse, sdk_httpx + +MODEL = "m" + + +async def _ignore(*_args, **_kwargs) -> None: + pass + + +class _Body(sdk_httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + + async def __aiter__(self): + for chunk in self.chunks: + yield chunk + + async def aclose(self) -> None: + pass + + +async def _run(cls, chunks: list[bytes], headers: dict[str, str] | None = None): + def respond(request): + return sdk_httpx.Response(200, stream=_Body(chunks), headers={ + "content-type": "text/event-stream", **(headers or {}), + }) + + client = cls(MODEL, api_key="k") + await client._client.close() + async with AsyncOpenAI(api_key="k", max_retries=0, http_client=sdk_httpx.AsyncClient( + transport=sdk_httpx.MockTransport(respond), + )) as sdk: + client._client = sdk + return await _stream_llm_response(client, [user_msg("hi")], 5, _ignore) + + +def _content(text: str = "done", finish: str | None = None) -> bytes: + return _sse(_chat_chunk(MODEL, {"content": text}, finish)) + + +def _split(data: bytes, size: int = 7) -> list[bytes]: + return [data[i:i + size] for i in range(0, len(data), size)] + + +def _deflate(data: bytes) -> bytes: + return zlib.compress(data) + + +# ── Chat Completions: complete turns ───────────────────────────────────── + +@pytest.mark.parametrize("encoding,compress", [ + ("gzip", gzip.compress), ("deflate", _deflate), +]) +async def test_encoded_stream_still_shows_done(encoding, compress): + """No finish_reason, only [DONE] — behind Content-Encoding.""" + body = compress(_content() + b"data: [DONE]\n\n") + response = await _run(OpenAIClient, _split(body), {"content-encoding": encoding}) + assert response.content == "done" + + +async def test_encoded_stream_without_done_is_still_truncated(): + body = gzip.compress(_content()) + with pytest.raises(LLMOpenAITruncatedStream): + await _run(OpenAIClient, [body], {"content-encoding": "gzip"}) + + +@pytest.mark.parametrize("done", [ + b"data: [DONE]\n\n", + b"data: [DONE]\r\n\r\n", + b"data: [DONE]\r\r", + b"data:[DONE]\n\n", + b"data: [DONE]", # body ends without the line break +], ids=["lf", "crlf", "cr", "no-space", "no-trailing-break"]) +async def test_done_line_endings(done): + response = await _run(OpenAIClient, [_content(), done]) + assert response.content == "done" + + +async def test_done_split_byte_by_byte(): + response = await _run(OpenAIClient, _split(_content() + b"data: [DONE]\n\n", 1)) + assert response.content == "done" + + +async def test_done_after_long_line_in_one_chunk(): + response = await _run(OpenAIClient, [_content("x" * 500) + b"data: [DONE]\n\n"]) + assert response.content == "x" * 500 + + +async def test_finish_reason_alone_completes_an_encoded_stream(): + body = gzip.compress(_content(finish="stop")) + response = await _run(OpenAIClient, [body], {"content-encoding": "gzip"}) + assert response.content == "done" + + +# ── Chat Completions: truncations ──────────────────────────────────────── + +@pytest.mark.parametrize("chunks", [ + [_content("par")], + [_sse(_chat_chunk(MODEL, {"tool_calls": [{ + "index": 0, "id": "c", "type": "function", + "function": {"name": "s", "arguments": '{"q'}}]}))], + [b""], + [_sse({"id": "x", "object": "chat.completion.chunk", "created": 1, "model": MODEL, + "choices": [], "usage": {"prompt_tokens": 1, "completion_tokens": 0, + "total_tokens": 1}})], + [_content("data: [DONE]\n")], # the sentinel inside model output + [_content(), b": data: [DONE]\n\n"], # inside an SSE comment + [_content(), b"event: x\ndata: [DONE"], # sentinel cut mid-token +], ids=["text", "mid-tool-call", "empty-body", "usage-only", "in-content", + "in-comment", "cut-sentinel"]) +async def test_chat_truncations_are_rejected(chunks): + with pytest.raises(LLMOpenAITruncatedStream) as info: + await _run(OpenAIClient, chunks) + assert classify_error(info.value) == "truncated_stream" + assert not is_transient_network(info.value) + + +async def test_chat_escape_hatch_accepts_and_logs(monkeypatch, caplog): + monkeypatch.setenv(STREAM_TERMINATOR_ENV, "0") + with caplog.at_level("WARNING"): + response = await _run(OpenAIClient, [_content("par")]) + assert response.content == "par" + assert "ended without [DONE] or finish_reason" in caplog.text + + +async def test_unobservable_stream_is_not_rejected(): + """A stream with no SDK decoder to watch: a missing finish_reason proves + nothing, so the turn is accepted rather than failed on every call.""" + + class _Iter: + def __init__(self): + self._chunks = iter([SimpleNamespace( + usage=None, model=MODEL, + choices=[SimpleNamespace( + delta=SimpleNamespace(content="done", tool_calls=None), + finish_reason=None, + )], + )]) + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._chunks) + except StopIteration: + raise StopAsyncIteration from None + + async def close(self): + pass + + client = OpenAIClient(MODEL, api_key="k") + + async def _open(_kwargs): + return _Iter() + + client._open_stream = _open + deltas = [d async for d in client.stream([user_msg("hi")])] + assert "".join(d.content for d in deltas) == "done" + + +# ── Responses ──────────────────────────────────────────────────────────── + +async def test_responses_error_event_keeps_its_message(): + chunks = [_sse({"type": "error", "code": "server_error", "message": "upstream boom", + "param": None, "sequence_number": 1}, "error")] + with pytest.raises(Exception) as info: + await _run(OpenAIResponsesClient, chunks) + assert not isinstance(info.value, LLMOpenAITruncatedStream) + assert "upstream boom" in str(info.value) + assert getattr(info.value, "status_code", None) == 500 + + +async def test_responses_escape_hatch(monkeypatch, caplog): + monkeypatch.setenv(STREAM_TERMINATOR_ENV, "0") + chunks = [_sse({"type": "response.output_text.delta", "item_id": "i", "output_index": 0, + "content_index": 0, "delta": "par", "sequence_number": 1}, + "response.output_text.delta")] + with caplog.at_level("WARNING"): + response = await _run(OpenAIResponsesClient, chunks) + assert response.content == "par" + assert "responses ended without" in caplog.text + + +# ── StreamActivity unit ────────────────────────────────────────────────── + +def test_long_lines_are_not_retained(): + activity = StreamActivity() + activity.observe_bytes(b"data: " + b"x" * 10_000) + assert activity._line == b"" and activity._line_long + activity.observe_bytes(b"\ndata: [DONE]\n") + assert activity.saw_done + + +# ── Retry routing ──────────────────────────────────────────────────────── + +class _TruncatingLLM: + """Raises a truncated stream ``truncate`` times, then completes.""" + + def __init__(self, truncate: int) -> None: + self.truncate = truncate + self.calls = 0 + + async def stream(self, messages, **_kw): + self.calls += 1 + yield StreamDelta(reasoning_content="thinking") + if self.calls <= self.truncate: + raise LLMTruncatedStream( + last_event="content_block_stop", events_seen=4, + saw_message_delta=False, block_types=["thinking"], elapsed_s=1.0, + ) + yield StreamDelta(content="done", finish_reason="stop") + + +async def test_no_chain_resamples_on_the_same_key(): + llm = _TruncatingLLM(truncate=2) + response = await call_llm( + llm, [user_msg("q")], timeout=30, max_retries=4, turn=1, + on_delta=_ignore, retry_wait_fixed=0, + ) + assert response.content == "done" + assert llm.calls == 3 + + +async def test_active_chain_advances_instead_of_resampling(): + llm = _TruncatingLLM(truncate=5) + with pytest.raises(LLMCallExhausted) as info: + await call_llm( + llm, [user_msg("q")], timeout=30, max_retries=4, turn=1, + on_delta=_ignore, retry_wait_fixed=0, + chain_fallback_active=lambda: True, + ) + assert info.value.reason == "chain_advance" + assert isinstance(info.value.last_exc, LLMTruncatedStream) + assert llm.calls == 1 + + +def test_openai_truncation_is_classified(): + err = LLMOpenAITruncatedStream(protocol="responses", last_event="", events_seen=0, + expected="response.completed", elapsed_s=0.0) + assert classify_error(err) == "truncated_stream"