diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 2eac6fc7b3..e44353d0e0 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -40,7 +40,7 @@ from agent.message_metadata import append_message, stamp_message_timestamp from agent.message_sanitization import ( _sanitize_surrogates, _repair_tool_call_arguments, normalize_finish_reason as _normalize_finish_reason, ) -from agent.reasoning_summaries import separate_glued_reasoning_blocks +from agent.reasoning_summaries import append_streamed_reasoning_detail, separate_glued_reasoning_blocks from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current from tools.terminal_tool_lifecycle import is_persistent_env from utils import base_url_host_matches, base_url_hostname, env_float, env_int @@ -2770,6 +2770,7 @@ class _StreamingCall(StreamingWaitMonitor): base_timeout, read_timeout, conn_cap = self._stream_timeouts() content_parts: list = [] reasoning_parts: list = [] + reasoning_details: list = [] # OpenRouter replay data (signatures, encrypted blocks) pending_text_parts: list[str] = [] tool_calls = _ToolCallAccumulator() tool_calls_acc = tool_calls.acc @@ -2844,6 +2845,15 @@ class _StreamingCall(StreamingWaitMonitor): reasoning_parts[-1] if reasoning_parts else "", reasoning_text) reasoning_parts.append(reasoning_text) self._emit_reasoning(reasoning_text) + # Structured reasoning_details deltas carry the provider's replay data; the + # non-streaming path already keeps them, so dropping them here lost + # reasoning continuity on nearly every turn. Pydantic parks unknown fields + # in ``model_extra``. + rd_delta = getattr(delta, "reasoning_details", None) + if rd_delta is None and isinstance(getattr(delta, "model_extra", None), dict): + rd_delta = delta.model_extra.get("reasoning_details") + for rd in rd_delta if isinstance(rd_delta, (list, tuple)) else (): + append_streamed_reasoning_detail(reasoning_details, rd) # Text (list-of-blocks deltas flattened once); possible echoed SSE is # buffered until it can be judged. @@ -2879,7 +2889,7 @@ class _StreamingCall(StreamingWaitMonitor): return self._adopt_final_response(stream.final_response) return self._finish_chat_stream(stream, role, content_parts, reasoning_parts, tool_calls_acc, finish_reason, model_name, usage_obj, flush_pending=_flush_pending_stream_text, - response_id=response_id, upstream_provider=upstream_provider) + response_id=response_id, upstream_provider=upstream_provider, reasoning_details=reasoning_details) def _adopt_final_response(self, final_response): """Adapter returned a completed response for ``stream=True``: switch the @@ -2928,7 +2938,7 @@ class _StreamingCall(StreamingWaitMonitor): return mock_tool_calls or None, has_truncated_tool_args def _finish_chat_stream(self, stream, role, content_parts, reasoning_parts, tool_calls_acc, finish_reason, - model_name, usage_obj, *, flush_pending, response_id=None, upstream_provider=None): + model_name, usage_obj, *, flush_pending, response_id=None, upstream_provider=None, reasoning_details=None): """Assemble the non-streaming-shaped response after the chunk loop. A stream ending with no finish_reason is a drop, not a completion: return a partial-stream stub so the loop fails fast instead of executing empty @@ -2965,6 +2975,10 @@ class _StreamingCall(StreamingWaitMonitor): raise provider_stream_error flush_pending() message = SimpleNamespace(role=role, content=full_content, tool_calls=mock_tool_calls, reasoning_content=full_reasoning) + if reasoning_details: + # Only when present: _build_assistant_message's passthrough persists them + # for replay, and non-reasoning providers keep the attribute absent. + message.reasoning_details = reasoning_details # The provider's id when the chunks carried one (chatcmpl-/gen-...): it is what a provider needs to # look a request up. Fabricated only when the stream never sent one. return SimpleNamespace(id=response_id or ("stream-" + str(uuid.uuid4())), model=model_name, usage=usage_obj, diff --git a/agent/reasoning_summaries.py b/agent/reasoning_summaries.py index ef6c7f7d89..72258ee48d 100644 --- a/agent/reasoning_summaries.py +++ b/agent/reasoning_summaries.py @@ -14,7 +14,7 @@ from typing import Any from agent.message_content import flatten_message_text -__all__ = ["separate_glued_reasoning_blocks"] +__all__ = ["append_streamed_reasoning_detail", "separate_glued_reasoning_blocks"] def separate_glued_reasoning_blocks(previous: str, delta: Any) -> str: @@ -29,3 +29,37 @@ def separate_glued_reasoning_blocks(previous: str, delta: Any) -> str: delta = flatten_message_text(delta, sep="") glued = previous and delta and not previous[-1].isspace() and delta.startswith("**") and "**" in delta[2:] return f"\n\n{delta}" if glued else delta + + +# reasoning_details entry types whose consecutive fragments are ONE logical block. +_MERGEABLE_DETAIL_TEXT_KEYS = {"reasoning.text": "text", "reasoning.summary": "summary"} +_BACKFILL_DETAIL_KEYS = ("signature", "id", "format", "index") + + +def append_streamed_reasoning_detail(details_acc: list, detail: Any) -> None: + """Accumulate one streamed ``reasoning_details`` delta entry into *details_acc*. + + OpenRouter streams ``reasoning_details`` as word-level deltas: consecutive + ``reasoning.text`` / ``reasoning.summary`` entries are fragments of one logical + block and are merged (later fragments backfill ``signature``/``id`` the first + omitted); encrypted/opaque entries stay discrete. Unmerged, a long thought + replays as hundreds of one-word entries and providers that validate the + sequence shape on the next turn reject it. SDK objects are normalized to dicts. + """ + if not isinstance(detail, dict): + if hasattr(detail, "model_dump"): + detail = detail.model_dump(warnings=False) + elif hasattr(detail, "__dict__"): + detail = dict(detail.__dict__) + else: + return + dtype = detail.get("type") + merge_key = _MERGEABLE_DETAIL_TEXT_KEYS.get(dtype) + last = details_acc[-1] if details_acc else None + if last is not None and merge_key and last.get("type") == dtype and isinstance(detail.get(merge_key), str): + last[merge_key] = (last.get(merge_key) or "") + detail[merge_key] + for k in _BACKFILL_DETAIL_KEYS: + if last.get(k) in (None, "") and detail.get(k) not in (None, ""): + last[k] = detail[k] + return + details_acc.append(dict(detail)) diff --git a/tests/agent/test_streamed_reasoning_details.py b/tests/agent/test_streamed_reasoning_details.py new file mode 100644 index 0000000000..4470630b87 --- /dev/null +++ b/tests/agent/test_streamed_reasoning_details.py @@ -0,0 +1,76 @@ +"""Streamed ``reasoning_details`` survive to the assembled response (replay continuity). + +The non-streaming path always kept OpenRouter's ``reasoning_details`` (signatures, +encrypted blocks a provider needs back on the next turn); the streaming chunk loop +dropped them. Consecutive text/summary fragments merge into one logical entry, +encrypted entries stay discrete. +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from agent.reasoning_summaries import append_streamed_reasoning_detail + + +def _make_chunk(content=None, finish_reason=None, model=None, reasoning_details=None, usage=None): + delta = SimpleNamespace(content=content, tool_calls=None, reasoning_content=None, reasoning=None) + if reasoning_details is not None: + delta.reasoning_details = reasoning_details + return SimpleNamespace(choices=[SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason)], + model=model, usage=usage) + + +def test_fragments_merge_per_block_and_backfill_signature(): + acc = [] + append_streamed_reasoning_detail(acc, {"type": "reasoning.text", "text": "The user "}) + append_streamed_reasoning_detail(acc, SimpleNamespace(type="reasoning.text", text="wants X.", signature="sig1")) + append_streamed_reasoning_detail(acc, {"type": "reasoning.encrypted", "data": "AAAA"}) + append_streamed_reasoning_detail(acc, {"type": "reasoning.encrypted", "data": "BBBB"}) + append_streamed_reasoning_detail(acc, {"type": "reasoning.summary", "summary": "s1 "}) + append_streamed_reasoning_detail(acc, {"type": "reasoning.summary", "summary": "s2"}) + assert [d["type"] for d in acc] == [ + "reasoning.text", "reasoning.encrypted", "reasoning.encrypted", "reasoning.summary"] + assert acc[0] == {"type": "reasoning.text", "text": "The user wants X.", "signature": "sig1"} + assert acc[3]["summary"] == "s1 s2" + + +def _agent(): + from run_agent import AIAgent + agent = AIAgent(api_key="test-key", base_url="https://openrouter.ai/api/v1", model="test/model", + quiet_mode=True, skip_context_files=True, skip_memory=True) + agent.api_mode = "chat_completions" + agent._interrupt_requested = False + return agent + + +@patch("run_agent.AIAgent._create_request_openai_client") +@patch("run_agent.AIAgent._close_request_openai_client") +def test_streamed_details_land_on_final_message_and_persist(_mock_close, mock_create): + chunks = [ + _make_chunk(reasoning_details=[{"type": "reasoning.text", "text": "I should "}]), + _make_chunk(reasoning_details=[{"type": "reasoning.text", "text": "answer.", "signature": "sigZ"}]), + _make_chunk(content="Hello!", finish_reason="stop", model="test-model"), + ] + mock_client = MagicMock() + mock_client.chat.completions.create.return_value = iter(chunks) + mock_create.return_value = mock_client + + agent = _agent() + response = agent._interruptible_streaming_api_call({}) + msg = response.choices[0].message + assert msg.content == "Hello!" + assert msg.reasoning_details == [{"type": "reasoning.text", "text": "I should answer.", "signature": "sigZ"}] + # The persisted assistant dict (what gets replayed next turn) carries them too. + persisted = agent._build_assistant_message(msg, "stop") + assert persisted["reasoning_details"] == msg.reasoning_details + + +@patch("run_agent.AIAgent._create_request_openai_client") +@patch("run_agent.AIAgent._close_request_openai_client") +def test_no_details_leaves_attribute_absent(_mock_close, mock_create): + mock_client = MagicMock() + mock_client.chat.completions.create.return_value = iter([ + _make_chunk(content="plain", finish_reason="stop", model="test-model")]) + mock_create.return_value = mock_client + response = _agent()._interruptible_streaming_api_call({}) + assert not hasattr(response.choices[0].message, "reasoning_details")