fix(agent): streamed reasoning_details survive for replay continuity (port of earendil-works/pi#8605)
The streaming accumulator dropped delta.reasoning_details entirely — only non-streaming responses preserved the OpenRouter unified reasoning replay data (signatures, encrypted reasoning blocks). Providers that require the reasoning_details sequence passed back on the next turn (OpenRouter reasoning models, Anthropic signed thinking via shims) lost continuity on every streamed turn, which is nearly all turns. - Accumulate delta.reasoning_details (attr or model_extra) during streaming and attach the merged list to the final mock message, where _build_assistant_message's existing passthrough persists it. - _append_streamed_reasoning_detail merges consecutive reasoning.text / reasoning.summary fragments into one logical entry (OpenRouter streams them as word-level deltas; unmerged they bloat the replayed signature payload) while encrypted/opaque entries stay discrete. Later fragments backfill signature/id/format/index fields the first fragment omitted. Ported from earendil-works/pi#8605 (commit c5ad7c1b0), credit @cristinaponcela for the merge-fragments pattern. Tests: tests/run_agent/test_streamed_reasoning_details.py (8 tests; sabotage-verified: disabling accumulation fails the E2E test, disabling merging fails 4 unit tests). Neighboring suites green (test_streaming.py 40 passed).
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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))
|
||||
|
||||
76
tests/agent/test_streamed_reasoning_details.py
Normal file
76
tests/agent/test_streamed_reasoning_details.py
Normal file
@@ -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")
|
||||
Reference in New Issue
Block a user