mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-05 02:41:56 +00:00
fix(anthropic): surface Responses bridge stream failures as Anthropic error events (#43126)
* fix(anthropic): surface Responses bridge stream failures as Anthropic error events The /v1/messages Responses bridge logged every upstream failure and ended the SSE stream as if it had completed, so a rate limit, a provider 500, a dropped connection, or a read timeout reached the client as HTTP 200 with a lone message_start and no error event. Map response.failed and any raised upstream exception to a redacted Anthropic error frame, stop pulling upstream after it, and never fabricate end_turn or message_stop after a failure. * fix(anthropic): close a Responses bridge stream that ends without a terminal event with an error event Normalize the failure status behind the error type to an int or digit string within 400..599, narrow the response.failed event through pydantic, reuse the native Messages path's incomplete-stream message for a clean upstream EOF, and cover the pydantic event, the unwrapped fallback error, and the EOF cases --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
3c93ea1697
commit
636eb4c396
5 changed files with 291 additions and 16 deletions
|
|
@ -16,6 +16,7 @@ from litellm.litellm_core_utils.core_helpers import process_response_headers
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.utils import INCOMPLETE_STREAM_ERROR_MESSAGE
|
||||
from litellm.proxy.pass_through_endpoints.success_handler import (
|
||||
PassThroughEndpointLogging,
|
||||
)
|
||||
|
|
@ -28,11 +29,6 @@ GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ: Final = PassThroughEndpointLogging()
|
|||
_UPSTREAM_PUMP_TASKS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: stdlib strong-ref set for pump tasks
|
||||
_DETACHED_STREAM_DRAINS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: bounded strong-ref set, detached drains
|
||||
|
||||
INCOMPLETE_STREAM_ERROR_MESSAGE: Final = (
|
||||
"Provider stream ended before emitting a message_stop event; "
|
||||
"the response is incomplete and any partial content (e.g. tool_use input JSON) may be truncated."
|
||||
)
|
||||
|
||||
|
||||
def _is_message_stop_chunk(chunk: object) -> bool:
|
||||
if isinstance(chunk, dict):
|
||||
|
|
|
|||
|
|
@ -15,6 +15,12 @@ if TYPE_CHECKING:
|
|||
from litellm.exceptions import ContentPolicyViolationError
|
||||
|
||||
|
||||
INCOMPLETE_STREAM_ERROR_MESSAGE: Final = (
|
||||
"Provider stream ended before emitting a message_stop event; "
|
||||
"the response is incomplete and any partial content (e.g. tool_use input JSON) may be truncated."
|
||||
)
|
||||
|
||||
|
||||
def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | None:
|
||||
"""
|
||||
Return the ``stop_details`` of an Anthropic Messages response refused by a
|
||||
|
|
|
|||
|
|
@ -2,20 +2,25 @@
|
|||
## Translates OpenAI call to Anthropic `/v1/messages` format
|
||||
import asyncio
|
||||
import json
|
||||
import traceback
|
||||
from collections import deque
|
||||
from collections.abc import AsyncIterator, Iterator, Mapping
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, field_validator
|
||||
|
||||
from litellm import verbose_logger
|
||||
from litellm._logging import redact_internal_details_from_client_message
|
||||
from litellm._uuid import uuid
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
encrypted_reasoning_signature,
|
||||
)
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
|
||||
INCOMPLETE_STREAM_ERROR_MESSAGE,
|
||||
refusal_stop_details,
|
||||
responses_output_refusal_text,
|
||||
)
|
||||
from litellm.responses.streaming_iterator import stream_error_status_and_message
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage
|
||||
|
||||
from .transformation import (
|
||||
|
|
@ -27,6 +32,72 @@ if TYPE_CHECKING:
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
|
||||
|
||||
|
||||
class _UpstreamFailure(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
status_code: int | None = None
|
||||
message: str | None = None
|
||||
|
||||
@field_validator("status_code", mode="before")
|
||||
@classmethod
|
||||
def http_error_status_or_none(cls, value: object) -> int | None:
|
||||
candidate: Final = (
|
||||
value
|
||||
if isinstance(value, int) and not isinstance(value, bool)
|
||||
else int(value)
|
||||
if isinstance(value, str) and value.isdecimal()
|
||||
else None
|
||||
)
|
||||
return candidate if candidate is not None and 400 <= candidate <= 599 else None
|
||||
|
||||
@field_validator("message", mode="before")
|
||||
@classmethod
|
||||
def str_or_none(cls, value: object) -> str | None:
|
||||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
class _FailedResponse(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, from_attributes=True)
|
||||
|
||||
error: object | None = None
|
||||
|
||||
|
||||
class _FailedResponseEvent(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, from_attributes=True)
|
||||
|
||||
response: _FailedResponse | None = None
|
||||
|
||||
|
||||
def _original_failure(exception: Exception) -> Exception:
|
||||
failure = exception # rebind-ok: walks the MidStreamFallbackError chain down to the provider failure
|
||||
while isinstance(failure, MidStreamFallbackError) and failure.original_exception is not None:
|
||||
failure = failure.original_exception
|
||||
return failure
|
||||
|
||||
|
||||
def _failure_status_and_message(exception: Exception) -> tuple[int, str]:
|
||||
original: Final = _original_failure(exception)
|
||||
failure: Final = _UpstreamFailure.model_validate(
|
||||
{"status_code": getattr(original, "status_code", None), "message": getattr(original, "message", None)}
|
||||
)
|
||||
status_code: Final = failure.status_code if failure.status_code is not None else 500
|
||||
message: Final = failure.message or str(original) or INCOMPLETE_STREAM_ERROR_MESSAGE
|
||||
return status_code, message
|
||||
|
||||
|
||||
def _anthropic_error_chunk(status_code: int, message: str) -> dict[str, object]:
|
||||
from litellm.anthropic_interface.exceptions.exception_mapping_utils import (
|
||||
AnthropicExceptionMapping,
|
||||
)
|
||||
|
||||
return dict(
|
||||
AnthropicExceptionMapping.transform_to_anthropic_error(
|
||||
status_code=status_code,
|
||||
raw_message=redact_internal_details_from_client_message(message),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class AnthropicResponsesStreamWrapper:
|
||||
"""
|
||||
Wraps a Responses API streaming iterator and re-emits events in Anthropic SSE format.
|
||||
|
|
@ -40,6 +111,7 @@ class AnthropicResponsesStreamWrapper:
|
|||
response.function_call_arguments.delta -> content_block_delta (input_json_delta)
|
||||
response.output_item.done -> content_block_delta (signature_delta) + content_block_stop
|
||||
response.completed -> message_delta + message_stop
|
||||
response.failed -> error (the stream ends without message_stop)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -60,6 +132,7 @@ class AnthropicResponsesStreamWrapper:
|
|||
self._pending_tool_ids: dict[str, str] = {} # item_id -> call_id / name accumulator
|
||||
self._sent_message_start = False
|
||||
self._sent_message_stop = False
|
||||
self._stream_failed = False
|
||||
self._chunk_queue: deque[dict[str, object]] = deque()
|
||||
self._refusal_text: str = ""
|
||||
self._sync_responses_iterator: Iterator[object] | None = None
|
||||
|
|
@ -293,10 +366,23 @@ class AnthropicResponsesStreamWrapper:
|
|||
)
|
||||
return
|
||||
|
||||
if event_type == "response.failed":
|
||||
failed: Final = _FailedResponseEvent.model_validate(event)
|
||||
status_code, message = stream_error_status_and_message(
|
||||
failed.response.error if failed.response is not None else None
|
||||
)
|
||||
verbose_logger.error(
|
||||
"AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed (%s): %s",
|
||||
self.model,
|
||||
status_code,
|
||||
message,
|
||||
)
|
||||
self._fail_stream(status_code, message)
|
||||
return
|
||||
|
||||
# ---- response completed -> message_delta + message_stop ----
|
||||
if event_type in (
|
||||
"response.completed",
|
||||
"response.failed",
|
||||
"response.incomplete",
|
||||
):
|
||||
response_obj: Final = getattr(event, "response", None) or (
|
||||
|
|
@ -350,21 +436,24 @@ class AnthropicResponsesStreamWrapper:
|
|||
self._sent_message_stop = True
|
||||
return
|
||||
|
||||
def _fail_stream(self, status_code: int, message: str) -> None:
|
||||
self._stream_failed = True
|
||||
self._chunk_queue.append(_anthropic_error_chunk(status_code, message))
|
||||
|
||||
def __aiter__(self) -> "AnthropicResponsesStreamWrapper":
|
||||
return self
|
||||
|
||||
async def __anext__(self) -> dict[str, object]:
|
||||
# Return any queued chunks first
|
||||
if self._chunk_queue:
|
||||
return self._chunk_queue.popleft()
|
||||
if self._stream_failed:
|
||||
raise StopAsyncIteration
|
||||
|
||||
# Emit message_start if not yet done (fallback if response.created wasn't fired)
|
||||
if not self._sent_message_start:
|
||||
self._sent_message_start = True
|
||||
self._chunk_queue.append(self._make_message_start())
|
||||
return self._chunk_queue.popleft()
|
||||
|
||||
# Consume the upstream stream
|
||||
try:
|
||||
if hasattr(self.responses_stream, "__aiter__"):
|
||||
async for event in self.responses_stream:
|
||||
|
|
@ -382,10 +471,19 @@ class AnthropicResponsesStreamWrapper:
|
|||
return self._chunk_queue.popleft()
|
||||
except StopAsyncIteration:
|
||||
pass
|
||||
except Exception as e:
|
||||
verbose_logger.error("AnthropicResponsesStreamWrapper error: %s\n%s", e, traceback.format_exc())
|
||||
except Exception as e: # noqa: BLE001 # every upstream failure becomes a client error event
|
||||
verbose_logger.exception(
|
||||
"AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed", self.model
|
||||
)
|
||||
self._fail_stream(*_failure_status_and_message(e))
|
||||
|
||||
if not self._chunk_queue and not self._sent_message_stop and not self._stream_failed:
|
||||
verbose_logger.error(
|
||||
"AnthropicResponsesStreamWrapper: upstream Responses stream for %s ended without a terminal event",
|
||||
self.model,
|
||||
)
|
||||
self._fail_stream(500, INCOMPLETE_STREAM_ERROR_MESSAGE)
|
||||
|
||||
# Drain any remaining queued chunks
|
||||
if self._chunk_queue:
|
||||
return self._chunk_queue.popleft()
|
||||
|
||||
|
|
|
|||
|
|
@ -230,6 +230,11 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None
|
|||
return next((status for status in map(_status_code_for_error_field, fields) if status is not None), 500)
|
||||
|
||||
|
||||
def stream_error_status_and_message(error_obj: object) -> tuple[int, str]:
|
||||
message, error_type, error_code = _error_event_fields(error_obj)
|
||||
return _status_code_for_error_fields(error_type, error_code), message
|
||||
|
||||
|
||||
def _map_stream_error_to_exception(error_obj: object, model: str, custom_llm_provider: str) -> Exception:
|
||||
from litellm.llms.base_llm.chat.transformation import BaseLLMException
|
||||
|
||||
|
|
|
|||
|
|
@ -4,18 +4,25 @@ Tests for AnthropicResponsesStreamWrapper
|
|||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
|
||||
|
||||
import litellm
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
encrypted_reasoning_signature,
|
||||
)
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.utils import INCOMPLETE_STREAM_ERROR_MESSAGE
|
||||
from litellm.llms.anthropic.experimental_pass_through.responses_adapters.streaming_iterator import (
|
||||
AnthropicResponsesStreamWrapper,
|
||||
)
|
||||
from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse
|
||||
|
||||
|
||||
def _process_all(events: list) -> list:
|
||||
|
|
@ -132,6 +139,7 @@ class TestReasoningItemWithoutSummaryText:
|
|||
{"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}},
|
||||
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hello"},
|
||||
{"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}},
|
||||
{"type": "response.completed"},
|
||||
]
|
||||
|
||||
def test_reasoning_without_summary_emits_no_thinking_block(self):
|
||||
|
|
@ -144,6 +152,8 @@ class TestReasoningItemWithoutSummaryText:
|
|||
("content_block_start", 0),
|
||||
("content_block_delta", 0),
|
||||
("content_block_stop", 0),
|
||||
("message_delta", None),
|
||||
("message_stop", None),
|
||||
]
|
||||
assert chunks[1]["content_block"] == {"type": "text", "text": ""}
|
||||
|
||||
|
|
@ -166,6 +176,8 @@ class TestReasoningItemWithoutSummaryText:
|
|||
("content_block_start", 1),
|
||||
("content_block_delta", 1),
|
||||
("content_block_stop", 1),
|
||||
("message_delta", None),
|
||||
("message_stop", None),
|
||||
]
|
||||
assert chunks[1]["content_block"] == {"type": "thinking", "thinking": "", "signature": ""}
|
||||
assert "".join(c["delta"]["thinking"] for c in chunks[2:4]) == "Weighing options"
|
||||
|
|
@ -215,6 +227,8 @@ class TestEncryptedReasoningIsStreamedForReplay:
|
|||
("content_block_start", 1),
|
||||
("content_block_delta", 1),
|
||||
("content_block_stop", 1),
|
||||
("message_delta", None),
|
||||
("message_stop", None),
|
||||
]
|
||||
assert chunks[1]["content_block"] == {
|
||||
"type": "redacted_thinking",
|
||||
|
|
@ -234,9 +248,7 @@ class TestEncryptedReasoningIsStreamedForReplay:
|
|||
]
|
||||
chunks = _process_all(events)
|
||||
|
||||
thinking = "".join(
|
||||
c["delta"]["thinking"] for c in chunks if c.get("delta", {}).get("type") == "thinking_delta"
|
||||
)
|
||||
thinking = "".join(c["delta"]["thinking"] for c in chunks if c.get("delta", {}).get("type") == "thinking_delta")
|
||||
assert thinking == "First.\n\nSecond."
|
||||
assert [c["type"] for c in chunks].count("content_block_start") == 1
|
||||
|
||||
|
|
@ -283,6 +295,7 @@ class TestToolUseBlockClosedExactlyOnce:
|
|||
"type": "response.output_item.done",
|
||||
"item": {"type": "message", "id": "chatcmpl-123", "status": "completed"},
|
||||
},
|
||||
{"type": "response.completed"},
|
||||
]
|
||||
|
||||
def test_one_content_block_stop_per_content_block_start(self):
|
||||
|
|
@ -302,6 +315,8 @@ class TestToolUseBlockClosedExactlyOnce:
|
|||
("content_block_delta", 0),
|
||||
("content_block_delta", 0),
|
||||
("content_block_stop", 0),
|
||||
("message_delta", None),
|
||||
("message_stop", None),
|
||||
]
|
||||
assert chunks[1]["content_block"] == {
|
||||
"type": "tool_use",
|
||||
|
|
@ -452,3 +467,158 @@ class TestRefusalStreamEvents:
|
|||
message_delta = next(c for c in chunks if c["type"] == "message_delta")
|
||||
assert message_delta["delta"]["stop_reason"] == "max_tokens"
|
||||
assert "stop_details" not in message_delta["delta"]
|
||||
|
||||
|
||||
def _collect(stream) -> list:
|
||||
async def _run() -> list:
|
||||
wrapper = AnthropicResponsesStreamWrapper(responses_stream=stream, model="m")
|
||||
return [chunk async for chunk in wrapper]
|
||||
|
||||
return asyncio.run(_run())
|
||||
|
||||
|
||||
class TestUpstreamFailureEndsStreamWithErrorEvent:
|
||||
"""A provider failure must reach the Anthropic client as an ``error`` event that
|
||||
ends the stream, never as a fabricated ``end_turn`` or a silent close."""
|
||||
|
||||
def test_response_failed_event_emits_error_event_and_stops_pulling_upstream(self):
|
||||
failed = SimpleNamespace(
|
||||
status="failed",
|
||||
output=[],
|
||||
usage=None,
|
||||
error={"code": "rate_limit_exceeded", "message": "Rate limit reached for gpt-5.5, try again in 20s."},
|
||||
)
|
||||
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
yield {"type": "response.failed", "response": failed}
|
||||
raise AssertionError("upstream was pulled again after the failure")
|
||||
|
||||
async def _run() -> list:
|
||||
wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m")
|
||||
return [frame async for frame in wrapper.async_anthropic_sse_wrapper()]
|
||||
|
||||
frames = asyncio.run(_run())
|
||||
assert [frame.split(b"\n", 1)[0] for frame in frames] == [b"event: message_start", b"event: error"]
|
||||
error_payload = json.loads(frames[1].split(b"data: ", 1)[1])
|
||||
assert error_payload["type"] == "error"
|
||||
assert error_payload["error"] == {
|
||||
"type": "rate_limit_error",
|
||||
"message": "Rate limit reached for gpt-5.5, try again in 20s.",
|
||||
}
|
||||
|
||||
def test_raised_mid_stream_fallback_error_is_unwrapped_to_the_provider_failure(self):
|
||||
rate_limit = litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m")
|
||||
wrapped = MidStreamFallbackError(
|
||||
message=str(rate_limit),
|
||||
model="m",
|
||||
llm_provider="openai",
|
||||
original_exception=rate_limit,
|
||||
is_pre_first_chunk=True,
|
||||
)
|
||||
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
raise wrapped
|
||||
|
||||
chunks = _collect(_gen())
|
||||
assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
|
||||
assert chunks[1]["error"] == {"type": "rate_limit_error", "message": rate_limit.message}
|
||||
|
||||
def test_sync_upstream_transport_error_after_content_becomes_api_error_event(self):
|
||||
def _events():
|
||||
yield {"type": "response.created"}
|
||||
yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}
|
||||
yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"}
|
||||
raise ConnectionResetError("Response payload is not completed")
|
||||
|
||||
chunks = _collect(_events())
|
||||
assert [chunk["type"] for chunk in chunks] == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"error",
|
||||
]
|
||||
assert chunks[-1]["error"] == {"type": "api_error", "message": "Response payload is not completed"}
|
||||
|
||||
def test_error_event_message_is_redacted_before_it_reaches_the_client(self):
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
raise RuntimeError("upstream failed with key sk-proj-abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJ")
|
||||
|
||||
chunks = _collect(_gen())
|
||||
assert chunks[-1]["type"] == "error"
|
||||
assert "sk-proj-" not in chunks[-1]["error"]["message"]
|
||||
assert chunks[-1]["error"]["message"].startswith("upstream failed with key")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raised", "expected_error"),
|
||||
[
|
||||
(
|
||||
MidStreamFallbackError(message="boom", model="m", llm_provider="openai"),
|
||||
{"type": "api_error", "message": "litellm.MidStreamFallbackError: boom"},
|
||||
),
|
||||
(
|
||||
type("StringStatusError", (Exception,), {"status_code": "429"})("throttled"),
|
||||
{"type": "rate_limit_error", "message": "throttled"},
|
||||
),
|
||||
(
|
||||
type("NonErrorStatusError", (Exception,), {"status_code": 200})("odd status"),
|
||||
{"type": "api_error", "message": "odd status"},
|
||||
),
|
||||
],
|
||||
ids=["mid-stream-fallback-without-original", "digit-string-status", "status-outside-4xx-5xx"],
|
||||
)
|
||||
def test_raised_failure_status_is_normalized_into_the_error_type(self, raised, expected_error):
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
raise raised
|
||||
|
||||
chunks = _collect(_gen())
|
||||
assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
|
||||
assert chunks[1]["error"] == expected_error
|
||||
|
||||
def test_pydantic_response_failed_event_is_mapped_like_a_dict_event(self):
|
||||
failed = ResponsesAPIResponse(
|
||||
id="resp_1",
|
||||
created_at=1,
|
||||
error={"code": "server_error", "message": "The server had an error while processing your request."},
|
||||
status="failed",
|
||||
output=[],
|
||||
model="m",
|
||||
object="response",
|
||||
parallel_tool_calls=False,
|
||||
tool_choice="auto",
|
||||
tools=[],
|
||||
)
|
||||
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
yield ResponseFailedEvent(type="response.failed", response=failed)
|
||||
|
||||
chunks = _collect(_gen())
|
||||
assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
|
||||
assert chunks[1]["error"] == {
|
||||
"type": "api_error",
|
||||
"message": "The server had an error while processing your request.",
|
||||
}
|
||||
|
||||
def test_upstream_ending_without_a_terminal_event_is_an_error_not_a_silent_close(self):
|
||||
async def _gen():
|
||||
yield {"type": "response.created"}
|
||||
yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}
|
||||
yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"}
|
||||
|
||||
chunks = _collect(_gen())
|
||||
assert [chunk["type"] for chunk in chunks] == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"error",
|
||||
]
|
||||
assert chunks[-1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}
|
||||
|
||||
def test_sync_upstream_ending_before_any_event_is_an_error_not_a_silent_close(self):
|
||||
chunks = _collect(iter(()))
|
||||
assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
|
||||
assert chunks[1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue