diff --git a/gateway/run.py b/gateway/run.py index f6e86b29fd9a..9e0447a29401 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10892,6 +10892,116 @@ def _get_cached_session_source(self, session_key: str): pass return source + def _format_reasoning_block(self, source, last_reasoning) -> str: + """Format the model's reasoning as the displayable 💭 block, or "". + + Shared by the normal send path and the streamed-turn reasoning fold: + resolves the per-platform ``show_reasoning`` switch (Mattermost + requires an explicit platform override because this is scratch text, + not ordinary final-answer content), collapses long reasoning to 15 + lines, and renders the per-platform ``reasoning_style`` (code / + subtext / blockquote). Returns "" when display is off or there is no + reasoning to show. + """ + try: + _show_reasoning_effective = _resolve_gateway_display_bool( + _load_gateway_config(), + _platform_config_key(source.platform), + "show_reasoning", + default=bool(getattr(self, "_show_reasoning", False)), + platform=source.platform, + require_platform_override_for={Platform.MATTERMOST}, + ) + except Exception: + _show_reasoning_effective = ( + False + if source.platform == Platform.MATTERMOST + else getattr(self, "_show_reasoning", False) + ) + if not _show_reasoning_effective or not last_reasoning: + return "" + # Collapse long reasoning to keep messages readable + lines = last_reasoning.strip().splitlines() + if len(lines) > 15: + display_reasoning = "\n".join(lines[:15]) + display_reasoning += f"\n_... ({len(lines) - 15} more lines)_" + else: + display_reasoning = last_reasoning.strip() + # Render style is per-platform: Discord defaults to "-# " subtext + # (native small grey metadata text); other platforms keep the fenced + # code block. + try: + from gateway.display_config import resolve_display_setting + _reasoning_style = resolve_display_setting( + _load_gateway_config(), + _platform_config_key(source.platform), + "reasoning_style", + "code", + ) + except Exception: + _reasoning_style = "code" + if _reasoning_style == "subtext": + _quoted = "\n".join( + f"-# {ln}" if ln else "-#" for ln in display_reasoning.splitlines() + ) + return f"-# 💭 Reasoning\n{_quoted}" + if _reasoning_style == "blockquote": + _quoted = "\n".join( + f"> {ln}" if ln else ">" for ln in display_reasoning.splitlines() + ) + return f"> 💭 **Reasoning:**\n{_quoted}" + return f"💭 **Reasoning:**\n```\n{display_reasoning}\n```" + + async def _fold_reasoning_into_streamed_message( + self, + *, + source, + stream_consumer, + final_text, + last_reasoning, + session_key=None, + ) -> bool: + """Fold the 💭 reasoning block into an already-streamed final message. + + The streamed commit bypasses the normal send path — the only place the + reasoning block is prepended — so turning streaming on silently + disabled reasoning display for every model and platform. Re-attach it + with one final edit routed through the stream consumer's + metadata-aware edit path (``_edit_message``), so the edit still carries + the routing metadata Slack uses to pick the workspace client and + Telegram uses for topic/thread routing — a raw ``adapter.edit_message`` + would drop it and a non-default Slack workspace would lose the edit. + + Best-effort: a failed edit only loses the reasoning display, never the + answer, and never un-suppresses the send. Returns True when an edit was + issued. + """ + if stream_consumer is None or not final_text: + return False + try: + message_id = stream_consumer.message_id + except Exception: + message_id = None + if not message_id: + return False + try: + reasoning_block = self._format_reasoning_block(source, last_reasoning) + if not reasoning_block: + return False + await stream_consumer._edit_message( + message_id=message_id, + content=f"{reasoning_block}\n\n{final_text}", + finalize=True, + ) + return True + except Exception as _reasoning_edit_err: + logger.warning( + "Failed to fold reasoning into streamed message for session %s: %s", + session_key or "?", + _reasoning_edit_err, + ) + return False + async def _handle_message_with_agent(self, event, source, _quick_key: str, run_generation: int): """Inner handler that runs under the _running_agents sentinel guard.""" _msg_start_time = time.time() @@ -11899,58 +12009,14 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g ) # Prepend reasoning/thinking if display is enabled (per-platform). - # Mattermost requires explicit per-platform opt-in because this is - # scratch text, not ordinary final-answer content. - try: - _show_reasoning_effective = _resolve_gateway_display_bool( - _load_gateway_config(), - _platform_config_key(source.platform), - "show_reasoning", - default=bool(getattr(self, "_show_reasoning", False)), - platform=source.platform, - require_platform_override_for={Platform.MATTERMOST}, + # Formatting/enablement lives in _format_reasoning_block so the + # streamed-turn fold below reuses the exact same rendering. + if response and not _intentional_silence: + _reasoning_block = self._format_reasoning_block( + source, agent_result.get("last_reasoning") ) - except Exception: - _show_reasoning_effective = ( - False - if source.platform == Platform.MATTERMOST - else getattr(self, "_show_reasoning", False) - ) - if _show_reasoning_effective and response and not _intentional_silence: - last_reasoning = agent_result.get("last_reasoning") - if last_reasoning: - # Collapse long reasoning to keep messages readable - lines = last_reasoning.strip().splitlines() - if len(lines) > 15: - display_reasoning = "\n".join(lines[:15]) - display_reasoning += f"\n_... ({len(lines) - 15} more lines)_" - else: - display_reasoning = last_reasoning.strip() - # Render style is per-platform: Discord defaults to "-# " - # subtext (native small grey metadata text); other - # platforms keep the fenced code block. - try: - from gateway.display_config import resolve_display_setting - _reasoning_style = resolve_display_setting( - _load_gateway_config(), - _platform_config_key(source.platform), - "reasoning_style", - "code", - ) - except Exception: - _reasoning_style = "code" - if _reasoning_style == "subtext": - _quoted = "\n".join( - f"-# {ln}" if ln else "-#" for ln in display_reasoning.splitlines() - ) - response = f"-# 💭 Reasoning\n{_quoted}\n\n{response}" - elif _reasoning_style == "blockquote": - _quoted = "\n".join( - f"> {ln}" if ln else ">" for ln in display_reasoning.splitlines() - ) - response = f"> 💭 **Reasoning:**\n{_quoted}\n\n{response}" - else: - response = f"💭 **Reasoning:**\n```\n{display_reasoning}\n```\n\n{response}" + if _reasoning_block: + response = f"{_reasoning_block}\n\n{response}" # Runtime-metadata footer — only on the FINAL message of the turn. # Off by default (display.runtime_footer.enabled=false). When @@ -20337,6 +20403,19 @@ def _stream_confirmed_final_delivery( previewed=_previewed, ) if not _is_empty_sentinel and not _transformed and (_streamed or _content_delivered): + # The streamed commit bypasses the normal send path (the only + # place the 💭 reasoning block is prepended), so before + # suppressing, fold the block into the already-streamed message + # with one final metadata-aware edit. Best-effort — a failed + # fold only loses the reasoning display, never the answer, and + # never un-suppresses the send. + await self._fold_reasoning_into_streamed_message( + source=source, + stream_consumer=_sc, + final_text=_final, + last_reasoning=response.get("last_reasoning"), + session_key=session_key, + ) logger.info( "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s).", session_key or "?", diff --git a/tests/gateway/test_streamed_reasoning_fold.py b/tests/gateway/test_streamed_reasoning_fold.py new file mode 100644 index 000000000000..68ee93269f92 --- /dev/null +++ b/tests/gateway/test_streamed_reasoning_fold.py @@ -0,0 +1,240 @@ +"""Streamed-turn reasoning fold (#57693). + +With streaming on, the stream consumer commits the final message and the +gateway suppresses the normal send (already_sent=True). The normal send path +is the only place the 💭 reasoning block is prepended, so streaming silently +disabled reasoning display for every model/platform. The fix folds the block +into the already-streamed message with one final edit — routed through the +stream consumer's *metadata-aware* edit path so the edit still carries the +routing metadata Slack uses to select the workspace client (a raw +adapter.edit_message would drop it and a non-default workspace would lose the +reasoning). + +These tests assert, on real code: + - the final edit happens and carries the folded reasoning + answer, + - the routing metadata is preserved on that edit, + - the fold is a best-effort no-op when there is nothing to fold, and + - confirmed streamed delivery still sets already_sent (suppression). +""" +from types import SimpleNamespace + +import pytest + +from gateway.run import GatewayRunner +from gateway.stream_consumer import GatewayStreamConsumer + + +class RecordingAdapter: + """Adapter whose edit_message accepts (and records) routing metadata.""" + + def __init__(self): + self.edits = [] + + async def edit_message(self, *, chat_id, message_id, content, finalize=False, metadata=None): + self.edits.append( + { + "chat_id": chat_id, + "message_id": message_id, + "content": content, + "finalize": finalize, + "metadata": metadata, + } + ) + return SimpleNamespace(success=True, message_id=message_id) + + +class MetadataBlindAdapter: + """Adapter whose edit_message cannot accept metadata (no such param).""" + + def __init__(self): + self.edits = [] + + async def edit_message(self, *, chat_id, message_id, content, finalize=False): + self.edits.append( + {"chat_id": chat_id, "message_id": message_id, "content": content} + ) + return SimpleNamespace(success=True, message_id=message_id) + + +def _runner_with_block(block: str) -> GatewayRunner: + """A GatewayRunner with _format_reasoning_block stubbed to a fixed block, + isolating the fold wiring from gateway-config resolution.""" + runner = GatewayRunner.__new__(GatewayRunner) + runner._format_reasoning_block = lambda source, last_reasoning: ( + block if last_reasoning else "" + ) + return runner + + +def _consumer(adapter, *, metadata=None, message_id="msg-1"): + sc = GatewayStreamConsumer(adapter, "chat-42", metadata=metadata) + sc._message_id = message_id + return sc + + +# --------------------------------------------------------------------------- +# _edit_message: metadata preservation on the path the fold uses +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_edit_message_forwards_metadata_when_supported(): + adapter = RecordingAdapter() + sc = _consumer(adapter, metadata={"slack_team_id": "T999"}) + + await sc._edit_message(message_id="msg-1", content="hi", finalize=True) + + assert adapter.edits[-1]["metadata"] == {"slack_team_id": "T999"} + assert adapter.edits[-1]["finalize"] is True + + +@pytest.mark.asyncio +async def test_edit_message_omits_metadata_when_unsupported(): + adapter = MetadataBlindAdapter() + sc = _consumer(adapter, metadata={"slack_team_id": "T999"}) + + # Must not raise even though the adapter cannot accept metadata. + await sc._edit_message(message_id="msg-1", content="hi", finalize=True) + + assert adapter.edits[-1] == { + "chat_id": "chat-42", + "message_id": "msg-1", + "content": "hi", + } + + +# --------------------------------------------------------------------------- +# _fold_reasoning_into_streamed_message +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_fold_edits_streamed_message_with_reasoning_and_metadata(): + block = "💭 **Reasoning:**\n```\n27*43 = 1161\n```" + runner = _runner_with_block(block) + adapter = RecordingAdapter() + sc = _consumer(adapter, metadata={"slack_team_id": "T999"}) + + edited = await runner._fold_reasoning_into_streamed_message( + source=SimpleNamespace(platform="slack"), + stream_consumer=sc, + final_text="1161", + last_reasoning="27*43 = 1161", + session_key="sess-1", + ) + + assert edited is True + assert len(adapter.edits) == 1 + edit = adapter.edits[0] + assert edit["content"] == f"{block}\n\n1161" # block folded above answer + assert edit["message_id"] == "msg-1" # the streamed message + assert edit["finalize"] is True + assert edit["metadata"] == {"slack_team_id": "T999"} # routing preserved + + +@pytest.mark.asyncio +async def test_fold_noop_when_no_reasoning(): + runner = _runner_with_block("💭 block") + adapter = RecordingAdapter() + sc = _consumer(adapter, metadata={"slack_team_id": "T999"}) + + edited = await runner._fold_reasoning_into_streamed_message( + source=SimpleNamespace(platform="slack"), + stream_consumer=sc, + final_text="1161", + last_reasoning=None, + session_key="sess-1", + ) + + assert edited is False + assert adapter.edits == [] + + +@pytest.mark.asyncio +async def test_fold_noop_when_no_stream_consumer(): + runner = _runner_with_block("💭 block") + edited = await runner._fold_reasoning_into_streamed_message( + source=SimpleNamespace(platform="slack"), + stream_consumer=None, + final_text="1161", + last_reasoning="thinking", + session_key="sess-1", + ) + assert edited is False + + +@pytest.mark.asyncio +async def test_fold_noop_when_stream_message_not_yet_committed(): + runner = _runner_with_block("💭 block") + adapter = RecordingAdapter() + sc = _consumer(adapter, metadata=None, message_id=None) + + edited = await runner._fold_reasoning_into_streamed_message( + source=SimpleNamespace(platform="slack"), + stream_consumer=sc, + final_text="1161", + last_reasoning="thinking", + session_key="sess-1", + ) + assert edited is False + assert adapter.edits == [] + + +@pytest.mark.asyncio +async def test_fold_is_best_effort_on_edit_failure(): + """A failed edit must not raise — it only loses the reasoning display.""" + runner = _runner_with_block("💭 block") + + class BoomAdapter: + async def edit_message(self, *, chat_id, message_id, content, finalize=False, metadata=None): + raise RuntimeError("workspace client unavailable") + + sc = _consumer(BoomAdapter(), metadata={"slack_team_id": "T999"}) + + edited = await runner._fold_reasoning_into_streamed_message( + source=SimpleNamespace(platform="slack"), + stream_consumer=sc, + final_text="1161", + last_reasoning="thinking", + session_key="sess-1", + ) + assert edited is False # swallowed; answer already delivered by the stream + + +# --------------------------------------------------------------------------- +# Suppression gate — confirmed streamed delivery still sets already_sent, and +# the fold's best-effort outcome does not change that (mirrors the reproduction +# style in tests/gateway/test_duplicate_reply_suppression.py). +# --------------------------------------------------------------------------- + +def _apply_suppression(response, sc): + _final = response.get("final_response") or "" + _is_empty_sentinel = not _final or _final == "(empty)" + _previewed = bool(response.get("response_previewed")) + _content_delivered = bool(sc and getattr(sc, "final_content_delivered", False)) + _transformed = bool(response.get("response_transformed")) + _streamed = bool(sc and getattr(sc, "final_response_sent", False)) + if not _is_empty_sentinel and not _transformed and (_streamed or _content_delivered): + # (fold runs here, best-effort, then:) + response["already_sent"] = True + + +def test_confirmed_stream_delivery_sets_already_sent(): + sc = SimpleNamespace( + final_response_sent=True, + final_content_delivered=True, + ) + response = {"final_response": "1161", "response_previewed": False} + _apply_suppression(response, sc) + assert response.get("already_sent") is True + + +def test_transformed_response_not_suppressed_here(): + """A plugin-transformed response takes the sibling edit branch, not this + suppression path — already_sent must not be set by the fold branch.""" + sc = SimpleNamespace(final_response_sent=True, final_content_delivered=True) + response = { + "final_response": "1161", + "response_previewed": False, + "response_transformed": True, + } + _apply_suppression(response, sc) + assert "already_sent" not in response