diff --git a/agent_core/components/middleware/llm/proxy.py b/agent_core/components/middleware/llm/proxy.py index 269c741..2e248b6 100644 --- a/agent_core/components/middleware/llm/proxy.py +++ b/agent_core/components/middleware/llm/proxy.py @@ -18,7 +18,7 @@ get_current_execution_scope, ) from agent_core.llm import LLMResponse, StreamDelta -from agent_core.messages import Message +from agent_core.messages import Message, ToolCall from agent_core.runtime.async_utils import await_bounded, closing_stream from agent_core.stream_tools import StreamToolCalls @@ -42,6 +42,7 @@ def __init__(self) -> None: self.stop_details: dict[str, Any] = {} self.stop_reason = "" self.tool_calls = StreamToolCalls() + self.final_tool_calls: list[ToolCall] | None = None def feed(self, delta: StreamDelta) -> None: # Last non-empty wins, except refusal text, which streams in pieces. @@ -61,6 +62,8 @@ def feed(self, delta: StreamDelta) -> None: self.stop_reason = delta.stop_reason # Keep known usage/provenance even if a malformed tool fragment fails. self.tool_calls.feed(delta.tool_call_deltas) + if delta.final_tool_calls is not None: + self.final_tool_calls = delta.final_tool_calls def response(self, content: str, reasoning: str) -> LLMResponse: metadata: dict[str, Any] = {} @@ -74,7 +77,8 @@ def response(self, content: str, reasoning: str) -> LLMResponse: metadata["stop_reason"] = self.stop_reason return LLMResponse( content=content, - tool_calls=self.tool_calls.complete(), + tool_calls=(self.final_tool_calls if self.final_tool_calls is not None + else self.tool_calls.complete()), reasoning_content=reasoning, finish_reason=self.finish_reason, model=self.model, diff --git a/agent_core/llm.py b/agent_core/llm.py index 03471d8..7ce58d3 100644 --- a/agent_core/llm.py +++ b/agent_core/llm.py @@ -84,6 +84,9 @@ class StreamDelta: usage_source: str = "" # Refusal text has its own SDK channel on OpenAI Chat and Responses. refusal: str = "" + # A provider's completed response may correct missing or divergent streamed + # fragments. None means no override; an empty list clears partial calls. + final_tool_calls: list[ToolCall] | None = None @runtime_checkable diff --git a/agent_core/providers/openai_responses.py b/agent_core/providers/openai_responses.py index d1da3e2..0cc9372 100644 --- a/agent_core/providers/openai_responses.py +++ b/agent_core/providers/openai_responses.py @@ -12,11 +12,12 @@ here (the Chat Completions endpoint 400s on that combo). Selected via a profile ``protocol: responses``. Reasoning is returned as a -typed block list, so the agent loop runs this client NON-STREAMING (see -``agent_loop`` protocol gate) — the verbatim reasoning items (incl. -``encrypted_content``) only survive intact on the non-streaming response, and +typed block list that :class:`~agent_core.runtime.loop.model_profile.DefaultThinkingParser` -(``content_block`` format) keeps them as ``raw_content_blocks``. +(``content_block`` format) keeps as ``raw_content_blocks``. ``chat`` and +``stream`` produce the same list — ``stream`` takes it from the terminal +``response.completed`` event — so the loop streams this protocol like any +other (``tests/test_responses_stream_replay.py``). LIVE-VERIFIED 2026-07-09 against a real Responses endpoint (gpt-5.5 via the miromind proxy): single-turn reasoning capture AND multi-turn @@ -184,11 +185,12 @@ async def stream( extra_headers: dict[str, str] | None = None, timeout: float | None = None, ) -> AsyncIterator[StreamDelta]: - # Provided for Protocol conformance. The agent loop forces this client - # non-streaming (reasoning blocks / encrypted_content only survive on - # the non-streaming response), so this path is not used to preserve - # verbatim reasoning — it maps the Responses semantic events to - # ``StreamDelta`` for any caller that streams anyway. + # Text / reasoning-summary / function-call argument deltas stream as + # they arrive; the terminal ``response`` is authoritative for replay. + # Its parsed block list (reasoning items with ``encrypted_content`` and + # function calls in provider order) rides on ``reasoning_blocks`` — + # exactly the ``content`` the non-streaming path returns — and any call + # whose events a gateway did not forward is emitted from it whole. kwargs = self._build_kwargs( messages, tools=tools, temperature=temperature, max_tokens=max_tokens, extra_headers=extra_headers, timeout=timeout, @@ -201,6 +203,8 @@ async def stream( events_seen = 0 last_event = "" saw_terminal = False + # output_index -> (call_id, name, arguments streamed so far) + streamed_calls: dict[int, tuple[str, str, str]] = {} async with contextlib.aclosing(stream_events_with_activity(stream, activity)) as events: async for event in events: if event is None: @@ -217,6 +221,29 @@ async def stream( refusal_seen = refusal_seen or bool(refusal) activity.mark_output() yield StreamDelta(content=refusal, refusal=refusal) + elif etype == "response.output_item.added": + item = getattr(event, "item", None) + if _get(item, "type", "") == "function_call": + index = int(getattr(event, "output_index", 0) or 0) + call_id = _get(item, "call_id", "") or _get(item, "id", "") or "" + args = _get(item, "arguments", "") or "" + name = _get(item, "name", "") or "" + streamed_calls[index] = (call_id, name, args) + activity.mark_output() + yield StreamDelta(tool_call_deltas=[{ + "index": index, "id": call_id, + "name": name, "arguments": args, + }]) + elif etype == "response.function_call_arguments.delta": + index = int(getattr(event, "output_index", 0) or 0) + chunk = getattr(event, "delta", "") or "" + if chunk: + call_id, name, args = streamed_calls.get(index, ("", "", "")) + streamed_calls[index] = (call_id, name, args + chunk) + activity.mark_output() + yield StreamDelta(tool_call_deltas=[{ + "index": index, "id": None, "name": None, "arguments": chunk, + }]) elif etype in ( "response.reasoning_summary_text.delta", "response.reasoning_text.delta", @@ -242,7 +269,10 @@ async def stream( refusal = parsed.response_metadata.get("refusal", "") if refusal and not refusal_seen: yield StreamDelta(content=refusal, refusal=refusal) - usage = _responses_usage_dict(getattr(resp, "usage", None)) + missing = _unstreamed_call_deltas(resp, streamed_calls) + if missing: + yield StreamDelta(tool_call_deltas=missing) + usage = _responses_usage_dict(_get(resp, "usage", None)) reason = normalize_finish_reason( _get(_get(resp, "incomplete_details", None) or {}, "reason", ""), ) @@ -250,9 +280,11 @@ async def stream( yield StreamDelta( usage=usage, stop_details=parsed.response_metadata.get("stop_details", {}), - usage_source="estimated" if usage.get("estimated") else "provider" if usage else "", - model=getattr(resp, "model", "") or "", + usage_source=parsed.usage_source, + model=parsed.model, finish_reason=reason or "stop", + reasoning_blocks=parsed.content if isinstance(parsed.content, list) else [], + final_tool_calls=parsed.tool_calls, ) if not saw_terminal: error = LLMOpenAITruncatedStream( @@ -267,6 +299,42 @@ async def stream( STREAM_TERMINATOR_ENV, error) +def _unstreamed_call_deltas( + resp: Any, streamed: dict[int, tuple[str, str, str]], +) -> list[dict[str, Any]]: + """Tool-call deltas that complete the streamed calls from the final response. + + A call the stream never opened is emitted whole; a call whose streamed + arguments are a prefix of the final ones gets the remainder. The terminal + ``final_tool_calls`` separately supplies the authoritative executable calls + when an append-only delta cannot correct divergent streamed arguments. + """ + out: list[dict[str, Any]] = [] + for index, item in enumerate(_get(resp, "output", None) or []): + if _get(item, "type", "") != "function_call": + continue + call_id = _get(item, "call_id", "") or _get(item, "id", "") or "" + final_args = _get(item, "arguments", "") or "{}" + if index not in streamed: + out.append({"index": index, "id": call_id, + "name": _get(item, "name", "") or "", "arguments": final_args}) + continue + seen_id, seen_name, partial = streamed[index] + missing_id = call_id if not seen_id else None + missing_name = (_get(item, "name", "") or "") if not seen_name else None + if final_args.startswith(partial): + if final_args != partial or missing_id or missing_name: + out.append({"index": index, "id": missing_id, "name": missing_name, + "arguments": final_args[len(partial):]}) + else: + logger.warning("Responses call %s streamed arguments diverge from the " + "completed response; using the completed call for execution", call_id) + if missing_id or missing_name: + out.append({"index": index, "id": missing_id, "name": missing_name, + "arguments": ""}) + return out + + # ── Conversion helpers (pure — unit-tested) ──────────────────────────────── diff --git a/agent_core/runtime/loop/_streaming.py b/agent_core/runtime/loop/_streaming.py index 22f1d51..9d0d8da 100644 --- a/agent_core/runtime/loop/_streaming.py +++ b/agent_core/runtime/loop/_streaming.py @@ -15,7 +15,7 @@ LLMStreamStalled, ) from agent_core.llm import LLMResponse -from agent_core.messages import Message +from agent_core.messages import Message, ToolCall from agent_core.runtime.async_utils import await_bounded from agent_core.stream_tools import StreamToolCalls @@ -245,6 +245,7 @@ async def _stream_llm_response( thinking_accum = "" delta_index = 0 tool_calls = StreamToolCalls() + final_tool_calls: list[ToolCall] | None = None # Terminal metadata streamed late by the provider — kept so the assembled # LLMResponse carries usage/finish_reason/model (else streaming runs report # 0 usage and observers never see finish_reason="length"). @@ -329,7 +330,7 @@ def _assembled_response() -> LLMResponse: # NAMED calls to tools that do not exist are deliberately kept: those # reach the executor and come back as "unknown tool 'x'", which the # model can read and act on. - complete_tool_calls = tool_calls.complete() + complete_tool_calls = final_tool_calls if final_tool_calls is not None else tool_calls.complete() # Warn rather than drop silently — a provider emitting these # consistently is a real upstream defect, and this is the only place # that can still see it. @@ -402,6 +403,8 @@ async def _close_chunk_stream() -> None: raw_visible = delta.content or getattr(delta, "refusal", "") or "" typed_thinking = delta.reasoning_content or "" tc_chunks = delta.tool_call_deltas or [] + if delta.final_tool_calls is not None: + final_tool_calls = delta.final_tool_calls # Capture terminal metadata as it arrives (usage on the # late ``include_usage`` chunk, finish_reason on the last # content chunk). Last non-empty wins. diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index 5a1e6a5..bbb9025 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -111,15 +111,18 @@ #: payload a ``redacted_thinking`` block carries with no deltas at all, and #: ``test_anthropic_latest_models.py`` runs the signature round-trip, the #: prefix-bound retry and the reset commit parametrized over streaming. -#: ``responses`` has one streaming test and it only covers error routing; -#: ``bedrock`` has none, and its transport is a different SDK client. +#: ``responses`` clears it too — ``stream`` hands back the completed response's +#: block list (``encrypted_content`` reasoning and function calls in provider +#: order), and ``test_responses_stream_replay.py`` requires a streamed tool turn +#: to produce the same history and follow-up request as the non-streaming one. +#: ``bedrock`` has no such test, and its transport is a different SDK client. #: #: The three were suppressed together and unconditionally when the loop engine #: was extracted (2026-09-01), one day BEFORE the provider substrate landed #: that block-list fidelity (2026-09-02) — so this was once correct for all #: three and nobody revisited it. Removing a protocol from this set is a matter #: of writing the test that proves its streamed replay. -UNVERIFIED_STREAM_PROTOCOLS = frozenset({"responses", "bedrock"}) +UNVERIFIED_STREAM_PROTOCOLS = frozenset({"bedrock"}) # Signature: (turn_index, messages_snapshot, metadata) -> awaitable None. diff --git a/changes/responses-stream-replay.fix.md b/changes/responses-stream-replay.fix.md new file mode 100644 index 0000000..0910441 --- /dev/null +++ b/changes/responses-stream-replay.fix.md @@ -0,0 +1 @@ +`OpenAIResponsesClient.stream` now streams function-call arguments and returns the completed response's verbatim block list (`encrypted_content` reasoning + function calls in provider order), so a streamed Responses turn replays exactly like the non-streaming one; previously streaming dropped every tool call and reasoning item. Argument deltas still provide progress when a gateway omits `response.output_item.added`, while the completed response controls tool execution if streamed fragments are missing or divergent. `responses` is removed from `UNVERIFIED_STREAM_PROTOCOLS`, so it now streams automatically when an observer or the reasoning-only watchdog asks for deltas. diff --git a/scripts/protocol_smoke/live_responses_stream_smoke.py b/scripts/protocol_smoke/live_responses_stream_smoke.py new file mode 100644 index 0000000..d93e694 --- /dev/null +++ b/scripts/protocol_smoke/live_responses_stream_smoke.py @@ -0,0 +1,119 @@ +"""Live smoke test: a real Responses-API turn, streamed vs non-streamed. + +Drives ``OpenAIResponsesClient`` through the real agent loop against an actual +llm-hub endpoint, once with ``stream_llm_tokens=False`` and once with +``True``, over a question that forces one tool call. Prints both final +answers and diffs the resulting message histories — the thing +``test_responses_stream_replay.py`` checks against a fake SDK, here checked +against the real wire format. + +Usage: + OPENAI_API_KEY=... uv run python scripts/protocol_smoke/live_responses_stream_smoke.py [model] + +Model defaults to gpt-4o (the only llm-hub model found that both supports +native ``/v1/responses`` and isn't blocked upstream as of 2026-10-14 — every +gpt-5.x/6.x reasoning model on that account 400s or returns a policy-violation +failure regardless of this fix; see temp/2026-10-14_responses-stream-live-smoke.md). +""" +from __future__ import annotations + +import asyncio +import json +import os +import sys + +from agent_core.loop_types import LoopConfig, LoopPolicy +from agent_core.providers.openai_responses import OpenAIResponsesClient +from agent_core.runtime.loop.agent_loop import run_agent_loop +from agent_core.runtime.loop.model_profile import ModelProfile + + +class Weather: + name = "get_weather" + + async def ainvoke(self, args: dict) -> str: + city = args.get("city", "") + return f"{city}: 14C, light rain" + + def to_openai_schema(self) -> dict: + return { + "type": "function", + "function": { + "name": self.name, + "description": "Get the current weather for a city", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + }, + } + + +async def _run(model: str, *, streaming: bool) -> tuple[str, list[dict]]: + base_url = (os.environ.get("LLM_HUB_BASE_URL") or os.environ.get("OPENAI_BASE_URL") + or "https://llm-hub.apodex.app/v1").removesuffix("/chat/completions") + api_key = os.environ.get("LLM_HUB_API_KEY") or os.environ.get("OPENAI_API_KEY") + client = OpenAIResponsesClient(model, api_key=api_key, base_url=base_url, timeout=90) + deltas: list[str] = [] + + class PrintObserver: + wants_llm_delta = streaming + + async def on_llm_delta(self, ctx) -> None: + text = getattr(ctx, "delta", "") or "" + if text: + deltas.append(text) + + result = await run_agent_loop( + system_prompt="You are a terse assistant. Use the weather tool when asked about weather.", + user_message="What's the weather in Berlin right now? Use the tool, then answer in one sentence.", + llm=client, tools=[Weather()], + config=LoopConfig( + max_turns=4, max_llm_retries=1, stream_llm_tokens=streaming, + loop_policy=LoopPolicy(no_tool_behavior="stop"), + ), + model_profile=ModelProfile( + model_id=model, provider="openai", protocol="responses", + thinking_format="content_block", + ), + observers=[PrintObserver()], + ) + print(f" [{'stream' if streaming else 'batch '}] deltas received: {len(deltas)}, " + f"joined len={sum(len(d) for d in deltas)}") + return result.final_content, result.messages + + +async def main() -> None: + model = sys.argv[1] if len(sys.argv) > 1 else "gpt-4o" + print(f"model={model}") + + print("-- non-streaming --") + batch_answer, batch_messages = await _run(model, streaming=False) + print("batch answer:", batch_answer) + + print("-- streaming --") + stream_answer, stream_messages = await _run(model, streaming=True) + print("stream answer:", stream_answer) + + print() + print("tool calls seen (batch): ", + [m.get("tool_calls") for m in batch_messages if m.get("role") == "assistant" and m.get("tool_calls")]) + print("tool calls seen (stream): ", + [m.get("tool_calls") for m in stream_messages if m.get("role") == "assistant" and m.get("tool_calls")]) + print() + same_shape = ( + len(batch_messages) == len(stream_messages) + and [m.get("role") for m in batch_messages] == [m.get("role") for m in stream_messages] + ) + print("same role sequence:", same_shape) + print() + print("batch messages:") + print(json.dumps(batch_messages, indent=2)[:3000]) + print() + print("stream messages:") + print(json.dumps(stream_messages, indent=2)[:3000]) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_responses_stream_replay.py b/tests/test_responses_stream_replay.py new file mode 100644 index 0000000..2a13646 --- /dev/null +++ b/tests/test_responses_stream_replay.py @@ -0,0 +1,188 @@ +"""A streamed Responses turn must replay exactly like its non-streaming twin. + +``responses`` sat in ``UNVERIFIED_STREAM_PROTOCOLS`` because nothing proved +that. It was not merely unproven: ``OpenAIResponsesClient.stream`` mapped only +text / reasoning-summary / usage events, so a streamed turn lost every +``function_call`` (the loop saw no tool call and stopped) and every +``encrypted_content`` reasoning item (the next request could not continue the +model's reasoning state). These tests drive the real client through the real +loop against a fake SDK, once per mode, and require the same history and the +same follow-up request. +""" +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import Any + +import pytest + +from agent_core.loop_types import LoopConfig, LoopPolicy +from agent_core.providers import openai_responses as rc +from agent_core.runtime.loop.agent_loop import run_agent_loop +from agent_core.runtime.loop.model_profile import ModelProfile + +USAGE = {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15} + +TOOL_TURN = { + "id": "resp_1", "model": "gpt-x", "status": "completed", "usage": USAGE, + "output": [ + {"type": "reasoning", "id": "rs_1", "encrypted_content": "ENC_A", + "summary": [{"type": "summary_text", "text": "look up a"}]}, + {"type": "function_call", "id": "fc_1", "call_id": "call_a", + "name": "search", "arguments": '{"q": "a"}'}, + {"type": "reasoning", "id": "rs_2", "encrypted_content": "ENC_B", "summary": []}, + {"type": "function_call", "id": "fc_2", "call_id": "call_b", + "name": "search", "arguments": '{"q": "b"}'}, + ], +} +FINAL_TURN = { + "id": "resp_2", "model": "gpt-x", "status": "completed", "usage": USAGE, + "output": [ + {"type": "reasoning", "id": "rs_3", "encrypted_content": "ENC_C", "summary": []}, + {"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "done"}]}, + ], +} + + +def _events(payload: dict[str, Any], *, item_events: bool, + drop_middle_chunk: bool = False): + """SDK-shaped semantic events for ``payload``, ending in ``completed``. + + ``item_events=False`` models a gateway that forwards only the deltas and + the terminal event — the completed response must still be enough. + """ + ev = SimpleNamespace + out: list[Any] = [] + for index, item in enumerate(payload["output"]): + if item_events: + added = {**item, "arguments": ""} if item["type"] == "function_call" else item + out.append(ev(type="response.output_item.added", output_index=index, item=added)) + if item["type"] == "reasoning": + for part in item["summary"]: + out.append(ev(type="response.reasoning_summary_text.delta", + output_index=index, delta=part["text"])) + elif item["type"] == "message": + for part in item["content"]: + text = part["text"] + for chunk in (text[:2], text[2:]): + out.append(ev(type="response.output_text.delta", + output_index=index, delta=chunk)) + elif item["type"] == "function_call": + args = item["arguments"] + chunks = (args[:5], args[5:-1], args[-1:]) + for chunk_index, chunk in enumerate(chunks): + if drop_middle_chunk and index == 1 and chunk_index == 1: + continue + out.append(ev(type="response.function_call_arguments.delta", + output_index=index, item_id=item["id"], delta=chunk)) + out.append(ev(type="response.completed", response=payload)) + + async def gen(): + for event in out: + yield event + return gen() + + +def _client(*, streaming: bool, item_events: bool = True, + drop_middle_chunk: bool = False): + requests: list[dict[str, Any]] = [] + + async def create(**kwargs): + requests.append(json.loads(json.dumps(kwargs["input"]))) + assert bool(kwargs.get("stream")) is streaming + payload = TOOL_TURN if len(requests) == 1 else FINAL_TURN + return (_events(payload, item_events=item_events, + drop_middle_chunk=drop_middle_chunk) if streaming else payload) + + client = rc.OpenAIResponsesClient("gpt-x", api_key="x") + client._client = SimpleNamespace(responses=SimpleNamespace(create=create)) + return client, requests + + +class Search: + name = "search" + + async def ainvoke(self, args): + return f"found {args.get('q')}" + + def to_openai_schema(self): + return {"type": "function", "function": { + "name": self.name, "parameters": {"type": "object"}, + }} + + +async def _run(*, streaming: bool, item_events: bool = True, + drop_middle_chunk: bool = False): + client, requests = _client(streaming=streaming, item_events=item_events, + drop_middle_chunk=drop_middle_chunk) + result = await run_agent_loop( + system_prompt="s", user_message="hi", llm=client, tools=[Search()], + config=LoopConfig( + max_turns=4, max_llm_retries=1, stream_llm_tokens=streaming, + loop_policy=LoopPolicy(no_tool_behavior="stop"), + ), + model_profile=ModelProfile( + model_id="gpt-x", provider="openai", protocol="responses", + thinking_format="content_block", + ), + ) + return result, requests + + +@pytest.mark.asyncio +@pytest.mark.parametrize("item_events", [True, False]) +@pytest.mark.parametrize("drop_middle_chunk", [True, False]) +async def test_streamed_tool_turn_replays_like_non_streaming(item_events, drop_middle_chunk): + batch, batch_requests = await _run(streaming=False) + streamed, streamed_requests = await _run( + streaming=True, item_events=item_events, drop_middle_chunk=drop_middle_chunk, + ) + + assert batch.final_content == streamed.final_content == "done" + assert len(streamed_requests) == len(batch_requests) == 2 + # The follow-up request is what the model actually sees: same items, same + # order, encrypted reasoning bound to the call that followed it. + assert streamed_requests[1] == batch_requests[1] + replay = [(i.get("type"), i.get("encrypted_content") or i.get("call_id")) + for i in streamed_requests[1] if i.get("type") in ("reasoning", "function_call")] + assert replay == [("reasoning", "ENC_A"), ("function_call", "call_a"), + ("reasoning", "ENC_B"), ("function_call", "call_b")] + assert json.dumps(streamed.messages, sort_keys=True) == json.dumps(batch.messages, sort_keys=True) + + +@pytest.mark.asyncio +async def test_stream_yields_each_call_once_with_full_arguments(): + client, _ = _client(streaming=True) + deltas = [d async for d in client.stream([{"role": "user", "content": "hi"}])] + from agent_core.stream_tools import StreamToolCalls + + calls = StreamToolCalls() + for d in deltas: + calls.feed(d.tool_call_deltas) + assert [(c["id"], c["function"]["arguments"]) for c in calls.complete()] == [ + ("call_a", '{"q": "a"}'), ("call_b", '{"q": "b"}'), + ] + assert deltas[-1].reasoning_blocks == rc._parse_responses_output(TOOL_TURN).content + + +@pytest.mark.asyncio +async def test_argument_deltas_arrive_without_item_added(): + client, _ = _client(streaming=True, item_events=False) + deltas = [d async for d in client.stream([{"role": "user", "content": "hi"}])] + arg_chunks = [part["arguments"] for delta in deltas[:-1] + for part in delta.tool_call_deltas if part.get("arguments")] + assert arg_chunks[:3] == ['{"q":', ' "a"', '}'] + assert deltas[-1].final_tool_calls == rc._parse_responses_output(TOOL_TURN).tool_calls + + +@pytest.mark.asyncio +async def test_terminal_calls_override_divergent_fragments_in_proxy(): + from agent_core.components.middleware.llm.proxy import _StreamTerminal + + client, _ = _client(streaming=True, drop_middle_chunk=True) + terminal = _StreamTerminal() + for delta in [d async for d in client.stream([{"role": "user", "content": "hi"}])]: + terminal.feed(delta) + assert terminal.response("", "").tool_calls == rc._parse_responses_output(TOOL_TURN).tool_calls diff --git a/tests/test_stream_llm_tokens_toggle.py b/tests/test_stream_llm_tokens_toggle.py index ac3f22c..64f9f5b 100644 --- a/tests/test_stream_llm_tokens_toggle.py +++ b/tests/test_stream_llm_tokens_toggle.py @@ -89,7 +89,7 @@ async def test_anthropic_streams_when_an_observer_wants_deltas() -> None: async def test_unverified_protocols_stay_non_streaming_but_say_so( protocol, caplog, ) -> None: - """Silence was the bug, not the suppression: these two have no test proving + """Silence was the bug, not the suppression: these have no test proving their streamed replay, so the default holds — and now announces itself.""" llm = RecordingLLM() with caplog.at_level("WARNING"): @@ -99,7 +99,7 @@ async def test_unverified_protocols_stay_non_streaming_but_say_so( @pytest.mark.asyncio -@pytest.mark.parametrize("protocol", ["anthropic", "bedrock", "chat_completions"]) +@pytest.mark.parametrize("protocol", ["anthropic", "bedrock", "chat_completions", "responses"]) async def test_explicit_true_streams_on_any_protocol(protocol) -> None: """The transport, not the observers, can be the reason to stream: a host that states it takes responsibility for its protocol's replay fidelity.""" @@ -142,8 +142,17 @@ async def test_reasoning_only_watchdog_also_selects_streaming() -> None: def test_the_verified_set_is_stated_not_guessed() -> None: """``anthropic`` leaving this set is the behaviour change; pin it so a future edit has to be deliberate.""" - assert frozenset({"responses", "bedrock"}) == UNVERIFIED_STREAM_PROTOCOLS + assert frozenset({"bedrock"}) == UNVERIFIED_STREAM_PROTOCOLS def test_new_transport_knob_preserves_positional_loop_config_arguments() -> None: assert inspect.signature(LoopConfig).parameters["stream_transport"].kind is inspect.Parameter.KEYWORD_ONLY + + +@pytest.mark.asyncio +async def test_responses_streams_when_an_observer_wants_deltas() -> None: + """``responses`` left the unverified set once its streamed replay was + proven (``test_responses_stream_replay.py``).""" + llm = RecordingLLM() + await _run(llm, protocol="responses", observers=[DeltaObserver()]) + assert (llm.stream_calls, llm.chat_calls) == (1, 0)