fix(providers): stream OpenAI Responses tool calls and reasoning, mark it verified

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,

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

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,17 +269,22 @@ 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", ""),
                     )
                     activity.mark_output()
                     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) ────────────────────────────────
 
 

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.

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.

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.

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())