From 2b127bdbbe26365136fcd323bc321d9e18fc18c1 Mon Sep 17 00:00:00 2001 From: Abhinavexist Date: Tue, 21 Jul 2026 05:26:49 +0530 Subject: [PATCH] =?UTF-8?q?fix:=20OpenAI=20compliance=20audit=20=E2=80=94?= =?UTF-8?q?=20.stream()=20events,=20streamed=20usage,=20request-id,=20esca?= =?UTF-8?q?pe=20hatches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - #1 .stream() now yields OpenAI ChatCompletionStreamEvents (content.delta/done, tool-call events) via ChatCompletionStreamState; / stripped from content events; text_deltas() for plain tokens; create(stream=True) unchanged. - #2 streamed usage + system_fingerprint captured on get_final_completion(). - #3 _request_id carried onto the returned completion. - #6 parse()/with_raw_response/with_streaming_response exposed on the wrapper (no more AttributeError; parse delegates to the OpenAI client, model default + guard). Rebased onto main (post #7–#11). --- README.md | 35 +++++-- pyproject.toml | 2 +- src/interfaze/_chat.py | 66 ++++++++++++- src/interfaze/_stream.py | 105 ++++++++++++-------- tests/test_compliance.py | 83 ++++++++++++++++ tests/test_stream.py | 206 ++++++++++++++++++++++++++++++--------- 6 files changed, 399 insertions(+), 98 deletions(-) create mode 100644 tests/test_compliance.py diff --git a/README.md b/README.md index 7363270..795b889 100644 --- a/README.md +++ b/README.md @@ -78,23 +78,22 @@ res = interfaze.chat.completions.create( ## Streaming -For live rendering, iterate `text_deltas()` — it yields visible text only, stripping Interfaze's -inline ``/`` side-channels: +`.stream()` yields OpenAI-style events (`content.delta`, `content.done`, tool-call events, …) with +Interfaze's inline ``/`` side-channels stripped from the content events: ```python stream = interfaze.chat.completions.stream( messages=[{"role": "user", "content": "Tell me a story."}], ) -for text in stream.text_deltas(): - print(text, end="") +for event in stream: + if event.type == "content.delta": + print(event.delta, end="") final = stream.get_final_completion() print(final.reasoning, final.precontext) ``` -> Iterating the stream directly (`for chunk in stream`) or the plain `create(stream=True)` path -> yields **raw** chunks whose `delta.content` still contains the ``/`` tags — use -> `text_deltas()` for anything user-facing. `.stream()` also accumulates and surfaces -> `reasoning`/`precontext` on `get_final_completion()`. +> Just want clean tokens? `stream.text_deltas()` yields visible text only. Plain +> `create(stream=True)` returns the raw chunk iterator (side-channel tags **not** stripped). ## Inputs @@ -152,6 +151,26 @@ precontext with `ChatInterfaze(precontext=[...])`. intermediary mid-job. - The underlying OpenAI client is available at `interfaze.openai`. +## Compatibility notes + +Drop-in for the OpenAI **chat completions** flow (`create`, response types, errors, +`create(stream=True)`, `models`), with a few behaviors worth knowing when migrating: + +- **`.stream()` yields OpenAI-style events**, drop-in with OpenAI's streaming helper — iterate + `event.type` (`"content.delta"` with `.delta`, `"content.done"`, + `"tool_calls.function.arguments.delta"`/`.done`, …), same as `client.chat.completions.stream()` + on the OpenAI SDK. `create(stream=True)` still gives the raw `ChatCompletionChunk` iterator, and + `stream.text_deltas()` gives just the clean visible text if you don't need events. +- **Returned text is lightly post-processed.** `json_object` content is unwrapped from its + ```` ```json ```` fence, and streamed ``/`` side-channels are pulled into + `reasoning`/`precontext`, so `message.content` may not be byte-identical to the raw wire response. +- **`inputs.*` accept https URLs** (Interfaze fetches them server-side). Those parts are valid for + Interfaze but **not** portable to OpenAI/Azure, which require base64 in `file`/`input_audio` parts. +- **Escape hatch:** anything not on the wrapper — `chat.completions.parse()`, `.with_raw_response`, + `.with_streaming_response` — is on the underlying client at `interfaze.openai`. +- **`tasks.*` return the extracted result** (a `dict`/`list`/`str`), not a `ChatCompletion` — e.g. + `tasks.ocr(...)` returns the OCR dict directly. + ## License MIT diff --git a/pyproject.toml b/pyproject.toml index 365f262..251edd3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,7 @@ dependencies = ["openai>=2,<3"] [project.urls] Homepage = "https://interfaze.ai" -Repository = "https://github.com/interfaze/interfaze-py" +Repository = "https://github.com/InterfazeAI/interfaze-python" [project.optional-dependencies] langchain = ["langchain-openai>=1.3.5; python_version>='3.10'"] diff --git a/src/interfaze/_chat.py b/src/interfaze/_chat.py index 92ce2c9..b55ba06 100644 --- a/src/interfaze/_chat.py +++ b/src/interfaze/_chat.py @@ -3,7 +3,7 @@ from typing import Any, Dict, Iterable, List, Literal, Optional, Union, cast, overload from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream -from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai.types.chat import ChatCompletion, ChatCompletionChunk, ParsedChatCompletion from ._constants import INTERFAZE_MODEL from ._errors import InterfazeError @@ -63,7 +63,11 @@ def to_interfaze(raw: ChatCompletion, strip_fence: bool) -> InterfazeChatComplet msg["content"] = strip_json_fence(msg["content"]) except (KeyError, IndexError, TypeError): pass - return InterfazeChatCompletion.model_validate(data) + result = InterfazeChatCompletion.model_validate(data) + request_id = getattr(raw, "_request_id", None) + if request_id is not None: + result._request_id = request_id + return result class _CompletionsBase: @@ -146,6 +150,35 @@ def stream( kw, strip = self._kwargs(messages, model, task, guard, response_format, kwargs) return InterfazeStream(self._client, kw, strip) + def parse( + self, + *, + messages: Iterable[Any], + response_format: Any, + model: str = INTERFAZE_MODEL, + guard: Optional[List[GuardCode]] = None, + **kwargs: Any, + ) -> "ParsedChatCompletion[Any]": + """Structured output via a Pydantic model (delegates to the OpenAI client's ``parse``). + + Interfaze extras (``vcache``/``precontext``) are present but untyped here; use ``create`` for + the typed extended completion. + """ + msgs = prepare(messages, model, None, guard, None)[0] + return self._client.chat.completions.parse( + model=model, messages=msgs, response_format=response_format, **kwargs + ) + + @property + def with_raw_response(self) -> Any: + """Raw-HTTP escape hatch (delegates to the OpenAI client; bypasses task/guard preprocessing).""" + return self._client.chat.completions.with_raw_response + + @property + def with_streaming_response(self) -> Any: + """Streaming raw-HTTP escape hatch (delegates to the OpenAI client).""" + return self._client.chat.completions.with_streaming_response + class AsyncCompletions(_CompletionsBase): """`interfaze.chat.completions` (async).""" @@ -211,6 +244,35 @@ def stream( kw, strip = self._kwargs(messages, model, task, guard, response_format, kwargs) return AsyncInterfazeStream(self._client, kw, strip) + async def parse( + self, + *, + messages: Iterable[Any], + response_format: Any, + model: str = INTERFAZE_MODEL, + guard: Optional[List[GuardCode]] = None, + **kwargs: Any, + ) -> "ParsedChatCompletion[Any]": + """Structured output via a Pydantic model (delegates to the OpenAI client's ``parse``). + + Interfaze extras (``vcache``/``precontext``) are present but untyped here; use ``create`` for + the typed extended completion. + """ + msgs = prepare(messages, model, None, guard, None)[0] + return await self._client.chat.completions.parse( + model=model, messages=msgs, response_format=response_format, **kwargs + ) + + @property + def with_raw_response(self) -> Any: + """Raw-HTTP escape hatch (delegates to the OpenAI client; bypasses task/guard preprocessing).""" + return self._client.chat.completions.with_raw_response + + @property + def with_streaming_response(self) -> Any: + """Streaming raw-HTTP escape hatch (delegates to the OpenAI client).""" + return self._client.chat.completions.with_streaming_response + class Chat: def __init__(self, client: OpenAI) -> None: diff --git a/src/interfaze/_stream.py b/src/interfaze/_stream.py index 542d033..cf14993 100644 --- a/src/interfaze/_stream.py +++ b/src/interfaze/_stream.py @@ -5,6 +5,7 @@ from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Tuple from openai import AsyncOpenAI, OpenAI +from openai.lib.streaming.chat import ChatCompletionStreamEvent, ChatCompletionStreamState from openai.types.chat import ChatCompletionChunk from ._errors import InterfazeError @@ -53,10 +54,7 @@ def _suffix_prefix_len(s: str, tag: str) -> int: class _SideChannelFilter: - """Strip inline ````/```` blocks from streamed content, chunk by chunk. - - Buffers a trailing partial that may be a split tag; never withholds text that cannot be a tag. - """ + """Incrementally strips ````/```` blocks, buffering tags split across chunks.""" def __init__(self) -> None: self._buf = "" @@ -102,6 +100,26 @@ def flush(self) -> str: return rest +def _clean_for_events( + chunk: ChatCompletionChunk, filt: _SideChannelFilter, role_injected: bool +) -> "Tuple[ChatCompletionChunk, bool]": + """Clean a chunk for openai's event state: strip side channels from content, inject a role.""" + cleaned = chunk.model_copy(deep=True) + if not cleaned.choices: + return cleaned, role_injected + delta = cleaned.choices[0].delta + if not role_injected: + role_injected = True + if not delta.role: + delta.role = "assistant" + piece = filt.feed(delta.content) if isinstance(delta.content, str) else "" + if cleaned.choices[0].finish_reason: + piece += filt.flush() + if isinstance(delta.content, str) or piece: + delta.content = piece + return cleaned, role_injected + + class _State: def __init__(self) -> None: self.content = "" @@ -111,6 +129,8 @@ def __init__(self) -> None: self.model = "" self.created = 0 self.tool_calls: Dict[int, Dict[str, str]] = {} + self.usage: Any = None + self.system_fingerprint: Optional[str] = None def accumulate(self, chunk: ChatCompletionChunk) -> None: if not self.id and chunk.id: @@ -119,6 +139,10 @@ def accumulate(self, chunk: ChatCompletionChunk) -> None: self.model = chunk.model if not self.created and chunk.created: self.created = chunk.created + if chunk.usage: + self.usage = chunk.usage + if chunk.system_fingerprint: + self.system_fingerprint = chunk.system_fingerprint if not chunk.choices: return choice = chunk.choices[0] @@ -159,6 +183,10 @@ def build(self, strip_fence: bool = False) -> InterfazeChatCompletion: ], "vcache": False, } + if self.usage is not None: + data["usage"] = self.usage.model_dump() + if self.system_fingerprint: + data["system_fingerprint"] = self.system_fingerprint if reasoning: data["reasoning"] = reasoning if precontext: @@ -167,13 +195,16 @@ def build(self, strip_fence: bool = False) -> InterfazeChatCompletion: class InterfazeStream: - """Sync streaming helper — iterate chunks, then ``get_final_completion()``.""" + """Sync streaming helper — iterate OpenAI-style events, then ``get_final_completion()``.""" def __init__(self, client: OpenAI, kwargs: Dict[str, Any], strip_fence: bool = False) -> None: self._client = client self._kwargs = kwargs self._strip_fence = strip_fence self._state = _State() + self._openai_state: ChatCompletionStreamState[None] = ChatCompletionStreamState() + self._filter = _SideChannelFilter() + self._role_injected = False self._started = False self._done = False @@ -183,36 +214,31 @@ def __enter__(self) -> "InterfazeStream": def __exit__(self, *exc: Any) -> None: return None - def __iter__(self) -> "Iterator[ChatCompletionChunk]": + def __iter__(self) -> "Iterator[ChatCompletionStreamEvent[None]]": if self._started: raise InterfazeError("This stream has already been consumed.") self._started = True for chunk in self._client.chat.completions.create(stream=True, **self._kwargs): self._state.accumulate(chunk) - yield chunk + cleaned, self._role_injected = _clean_for_events(chunk, self._filter, self._role_injected) + yield from self._openai_state.handle_chunk(cleaned) self._done = True def text_deltas(self) -> "Iterator[str]": - """Yield visible text only, stripping ````/```` across chunk boundaries. - - Use this (not raw ``create(stream=True)`` deltas) for live rendering. ``reasoning`` and - ``precontext`` remain available on ``get_final_completion()``. - """ + """Iterate only the visible text (no side channels, no events) — for plain-token consumers.""" if self._started: raise InterfazeError("This stream has already been consumed.") self._started = True - filt = _SideChannelFilter() for chunk in self._client.chat.completions.create(stream=True, **self._kwargs): self._state.accumulate(chunk) - if chunk.choices: - delta = chunk.choices[0].delta - if delta and isinstance(delta.content, str) and delta.content: - visible = filt.feed(delta.content) - if visible: - yield visible - tail = filt.flush() - if tail: - yield tail + if not chunk.choices: + continue + delta = chunk.choices[0].delta + piece = self._filter.feed(delta.content) if isinstance(delta.content, str) else "" + if chunk.choices[0].finish_reason: + piece += self._filter.flush() + if piece: + yield piece self._done = True @property @@ -234,13 +260,16 @@ def get_final_completion(self) -> InterfazeChatCompletion: class AsyncInterfazeStream: - """Async streaming helper — ``async for`` chunks, then ``await get_final_completion()``.""" + """Async streaming helper — ``async for`` OpenAI-style events, then ``await get_final_completion()``.""" def __init__(self, client: AsyncOpenAI, kwargs: Dict[str, Any], strip_fence: bool = False) -> None: self._client = client self._kwargs = kwargs self._strip_fence = strip_fence self._state = _State() + self._openai_state: ChatCompletionStreamState[None] = ChatCompletionStreamState() + self._filter = _SideChannelFilter() + self._role_injected = False self._started = False self._done = False @@ -250,38 +279,34 @@ async def __aenter__(self) -> "AsyncInterfazeStream": async def __aexit__(self, *exc: Any) -> None: return None - async def __aiter__(self) -> "AsyncIterator[ChatCompletionChunk]": + async def __aiter__(self) -> "AsyncIterator[ChatCompletionStreamEvent[None]]": if self._started: raise InterfazeError("This stream has already been consumed.") self._started = True stream = await self._client.chat.completions.create(stream=True, **self._kwargs) async for chunk in stream: self._state.accumulate(chunk) - yield chunk + cleaned, self._role_injected = _clean_for_events(chunk, self._filter, self._role_injected) + for event in self._openai_state.handle_chunk(cleaned): + yield event self._done = True async def text_deltas(self) -> "AsyncIterator[str]": - """Yield visible text only, stripping ````/```` across chunk boundaries. - - Use this (not raw ``create(stream=True)`` deltas) for live rendering. ``reasoning`` and - ``precontext`` remain available on ``get_final_completion()``. - """ + """Iterate only the visible text (no side channels, no events) — for plain-token consumers.""" if self._started: raise InterfazeError("This stream has already been consumed.") self._started = True - filt = _SideChannelFilter() stream = await self._client.chat.completions.create(stream=True, **self._kwargs) async for chunk in stream: self._state.accumulate(chunk) - if chunk.choices: - delta = chunk.choices[0].delta - if delta and isinstance(delta.content, str) and delta.content: - visible = filt.feed(delta.content) - if visible: - yield visible - tail = filt.flush() - if tail: - yield tail + if not chunk.choices: + continue + delta = chunk.choices[0].delta + piece = self._filter.feed(delta.content) if isinstance(delta.content, str) else "" + if chunk.choices[0].finish_reason: + piece += self._filter.flush() + if piece: + yield piece self._done = True async def get_final_completion(self) -> InterfazeChatCompletion: diff --git a/tests/test_compliance.py b/tests/test_compliance.py new file mode 100644 index 0000000..78419f0 --- /dev/null +++ b/tests/test_compliance.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import asyncio + +import httpx +import respx +from conftest import BASIC, CHAT_URL, _chunk, completion, mock_json, mock_sse +from pydantic import BaseModel + +from interfaze import AsyncInterfaze, Interfaze + +USAGE_CHUNK = { + "id": "req-test", + "object": "chat.completion.chunk", + "created": 1_700_000_000, + "model": "interfaze-beta", + "choices": [], + "usage": {"prompt_tokens": 216, "completion_tokens": 10, "total_tokens": 226}, + "system_fingerprint": "fp_test", +} +STREAM_WITH_USAGE = [_chunk({"content": "Hi"}), _chunk({}, finish_reason="stop"), USAGE_CHUNK] + + +@respx.mock +def test_stream_captures_usage_and_fingerprint(): + mock_sse(STREAM_WITH_USAGE) + stream = Interfaze(api_key="t").chat.completions.stream( + messages=[{"role": "user", "content": "x"}], stream_options={"include_usage": True} + ) + final = stream.get_final_completion() + assert final.usage is not None + assert final.usage.prompt_tokens == 216 + assert final.usage.completion_tokens == 10 + assert final.usage.total_tokens == 226 + assert final.system_fingerprint == "fp_test" + + +@respx.mock +def test_async_stream_captures_usage(): + mock_sse(STREAM_WITH_USAGE) + + async def go(): + stream = AsyncInterfaze(api_key="t").chat.completions.stream( + messages=[{"role": "user", "content": "x"}], stream_options={"include_usage": True} + ) + async for _ in stream: + pass + return await stream.get_final_completion() + + final = asyncio.run(go()) + assert final.usage is not None and final.usage.total_tokens == 226 + + +@respx.mock +def test_request_id_carried_through(): + respx.post(CHAT_URL).mock( + return_value=httpx.Response(200, json=BASIC, headers={"x-request-id": "req-abc123"}) + ) + r = Interfaze(api_key="t").chat.completions.create(messages=[{"role": "user", "content": "x"}]) + assert r._request_id == "req-abc123" + + +class _Weather(BaseModel): + city: str + temp_c: float + + +@respx.mock +def test_parse_returns_parsed_object(): + mock_json(completion('{"city": "Tokyo", "temp_c": 21}')) + res = Interfaze(api_key="t").chat.completions.parse( + messages=[{"role": "user", "content": "Weather in Tokyo?"}], + response_format=_Weather, + ) + parsed = res.choices[0].message.parsed + assert isinstance(parsed, _Weather) and parsed.city == "Tokyo" and parsed.temp_c == 21 + + +def test_escape_hatch_surface_present(): + for comp in (Interfaze(api_key="t").chat.completions, AsyncInterfaze(api_key="t").chat.completions): + assert hasattr(comp, "parse") + assert comp.with_raw_response is not None + assert comp.with_streaming_response is not None diff --git a/tests/test_stream.py b/tests/test_stream.py index 6bb7d98..5327e5e 100644 --- a/tests/test_stream.py +++ b/tests/test_stream.py @@ -7,6 +7,36 @@ from conftest import STREAM_CHUNKS, STREAM_THINK, _chunk, mock_sse from interfaze import AsyncInterfaze, Interfaze +from interfaze._errors import InterfazeError + +TOOL_CALL_CHUNKS = [ + _chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": ""}, + } + ] + } + ), + _chunk({"tool_calls": [{"index": 0, "function": {"arguments": '{"city":'}}]}), + _chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"Paris"}'}}]}), + _chunk({}, finish_reason="tool_calls"), +] + + +def _content_deltas(events) -> str: + return "".join(e.delta for e in events if e.type == "content.delta") + + +def _assert_no_side_channel_leak(events) -> None: + for e in events: + text = str(e) + assert "" not in text and "" not in text + FENCED_JSON = [ _chunk({"content": "```json\n"}), @@ -17,11 +47,20 @@ @respx.mock -def test_stream_iterates_and_accumulates(): +def test_stream_events_strip_precontext(): mock_sse(STREAM_CHUNKS) stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) - chunks = list(stream) - assert len(chunks) == len(STREAM_CHUNKS) + events = list(stream) + _assert_no_side_channel_leak(events) + + joined = _content_deltas(events) + assert "" not in joined + assert "$12.34" in joined + + done = [e for e in events if e.type == "content.done"] + assert len(done) == 1 + assert "" not in done[0].content + final = stream.get_final_completion() content = final.choices[0].message.content or "" assert "" not in content and "$12.34" in content @@ -30,75 +69,125 @@ def test_stream_iterates_and_accumulates(): @respx.mock -def test_stream_think_parsed_without_iterating(): +def test_stream_events_strip_think(): mock_sse(STREAM_THINK) stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) - final = stream.get_final_completion() # consumes internally + events = list(stream) + _assert_no_side_channel_leak(events) + + joined = _content_deltas(events) + assert "" not in joined + assert "The sky is blue." in joined + + final = stream.get_final_completion() assert final.reasoning and "Rayleigh" in final.reasoning assert "" not in (final.choices[0].message.content or "") @respx.mock -def test_async_stream(): - mock_sse(STREAM_CHUNKS) +def test_stream_get_final_completion_without_iterating(): + mock_sse(STREAM_THINK) + stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) + final = stream.get_final_completion() # consumes internally, no events ever built + assert final.reasoning and "Rayleigh" in final.reasoning + assert "" not in (final.choices[0].message.content or "") - async def go(): - stream = AsyncInterfaze(api_key="t").chat.completions.stream( - messages=[{"role": "user", "content": "x"}] - ) - n = 0 - async for _ in stream: - n += 1 - return n, await stream.get_final_completion() - n, final = asyncio.run(go()) - assert n == len(STREAM_CHUNKS) - assert final.precontext and final.precontext[0].name == "ocr" +@respx.mock +def test_split_tag_across_chunks_no_leak_no_drop(): + chunks = [ + _chunk({"content": "x[{"name":"ocr","result":{}}]y'}), + _chunk({}, finish_reason="stop"), + ] + mock_sse(chunks) + stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) + events = list(stream) + _assert_no_side_channel_leak(events) + assert _content_deltas(events) == "xy" @respx.mock -def test_text_deltas_strips_precontext(): - mock_sse(STREAM_CHUNKS) +def test_eof_with_unresolved_tag_prefix_is_flushed_not_dropped(): + # An unterminated "" not in text + events = list(stream) + assert _content_deltas(events) == "Hello " not in text + events = list(stream) + assert _content_deltas(events) == "a < b" @respx.mock -def test_text_deltas_handles_tag_split_across_chunks(): - chunks = [ - _chunk({"content": "Hello [{"name":"ocr","result":1}] world"}), - _chunk({}, finish_reason="stop"), - ] - mock_sse(chunks) +def test_tool_call_stream_events_and_final(): + mock_sse(TOOL_CALL_CHUNKS) + stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) + events = list(stream) + + done = [e for e in events if e.type == "tool_calls.function.arguments.done"] + assert len(done) == 1 + assert done[0].name == "get_weather" + assert done[0].arguments == '{"city":"Paris"}' + assert not any(e.type == "content.done" for e in events) + + final = stream.get_final_completion() + assert final.choices[0].finish_reason == "tool_calls" + assert final.choices[0].message.content is None + tool_calls = final.choices[0].message.tool_calls + assert tool_calls and tool_calls[0].function.name == "get_weather" + assert tool_calls[0].function.arguments == '{"city":"Paris"}' + + +@respx.mock +def test_text_deltas_sync(): + mock_sse(STREAM_CHUNKS) stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) text = "".join(stream.text_deltas()) - assert "precontext" not in text and "ocr" not in text - assert text == "Hello world" + assert "" not in text + assert "$12.34" in text @respx.mock -def test_text_deltas_preserves_literal_lt(): - chunks = [ - _chunk({"content": "a < b and c "}), - _chunk({"content": "< d"}), - _chunk({}, finish_reason="stop"), - ] - mock_sse(chunks) +def test_double_consume_raises(): + mock_sse(STREAM_CHUNKS) stream = Interfaze(api_key="t").chat.completions.stream(messages=[{"role": "user", "content": "x"}]) - assert "".join(stream.text_deltas()) == "a < b and c < d" + list(stream) + try: + list(stream) + assert False, "expected InterfazeError" + except InterfazeError: + pass + + +@respx.mock +def test_async_stream_events(): + mock_sse(STREAM_CHUNKS) + + async def go(): + stream = AsyncInterfaze(api_key="t").chat.completions.stream( + messages=[{"role": "user", "content": "x"}] + ) + events = [] + async for e in stream: + events.append(e) + return events, await stream.get_final_completion() + + events, final = asyncio.run(go()) + _assert_no_side_channel_leak(events) + joined = _content_deltas(events) + assert "" not in joined and "$12.34" in joined + assert final.precontext and final.precontext[0].name == "ocr" @respx.mock @@ -109,7 +198,10 @@ async def go(): stream = AsyncInterfaze(api_key="t").chat.completions.stream( messages=[{"role": "user", "content": "x"}] ) - return "".join([t async for t in stream.text_deltas()]) + pieces = [] + async for piece in stream.text_deltas(): + pieces.append(piece) + return "".join(pieces) assert asyncio.run(go()) == "Total is $12.34" @@ -156,3 +248,23 @@ async def go(): content = asyncio.run(go()).choices[0].message.content or "" assert not content.lstrip().startswith("```") + + +@respx.mock +def test_async_double_consume_raises(): + mock_sse(STREAM_CHUNKS) + + async def go(): + stream = AsyncInterfaze(api_key="t").chat.completions.stream( + messages=[{"role": "user", "content": "x"}] + ) + async for _ in stream: + pass + try: + async for _ in stream: + pass + return False + except InterfazeError: + return True + + assert asyncio.run(go())