diff --git a/api/streaming.py b/api/streaming.py index e18e8792..7973658a 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -3804,6 +3804,18 @@ def _run_agent_streaming( stats.setdefault('estimated', False) put('metering', stats) + def _compact_for_echo_compare(value: str) -> str: + return re.sub(r'\s+', ' ', str(value or '')).strip() + + def _is_visible_output_echo(text: str) -> bool: + candidate = _compact_for_echo_compare(text) + if not candidate: + return False + visible_tail = _compact_for_echo_compare( + STREAM_PARTIAL_TEXT.get(stream_id, '')[-max(len(str(text)) * 2, 512):] + ) + return bool(visible_tail and visible_tail.endswith(candidate)) + def on_token(text): nonlocal _token_sent if text is None: @@ -3824,11 +3836,18 @@ def _run_agent_streaming( nonlocal _reasoning_text if text is None: return - _reasoning_text += str(text) + reasoning_delta = str(text) + # Some runtimes mirror user-visible progress text through the + # reasoning channel after it already streamed as normal assistant + # output. Treat that as an echo, otherwise the UI renders the + # same sentence again inside a Thinking card. + if _is_visible_output_echo(reasoning_delta): + return + _reasoning_text += reasoning_delta # Mirror to shared dict so cancel_stream() can persist it (#1361 §A) if stream_id in STREAM_REASONING_TEXT: - STREAM_REASONING_TEXT[stream_id] += str(text) - put('reasoning', {'text': str(text)}) + STREAM_REASONING_TEXT[stream_id] += reasoning_delta + put('reasoning', {'text': reasoning_delta}) # Track reasoning deltas in the meter so live TPS reflects all AI output. _metering_reasoning_deltas[0] += 1 meter().record_reasoning(stream_id, _metering_reasoning_deltas[0]) @@ -3840,9 +3859,10 @@ def _run_agent_streaming( visible = str(text).strip() if not visible: return + already_streamed = bool(cb_kwargs.get('already_streamed', False)) or _is_visible_output_echo(visible) put('interim_assistant', { 'text': visible, - 'already_streamed': bool(cb_kwargs.get('already_streamed', False)), + 'already_streamed': already_streamed, }) # Pre-initialise the activity counter here so on_tool (which @@ -3912,11 +3932,17 @@ def _run_agent_streaming( if event_type in ('reasoning.available', '_thinking'): reason_text = preview if event_type == 'reasoning.available' else name if reason_text: - _reasoning_text += str(reason_text) + reason_delta = str(reason_text) + # Older tool-progress paths can mirror the same visible + # progress text already emitted through stream_delta_callback. + # Suppress those echoes like the dedicated reasoning callback. + if _is_visible_output_echo(reason_delta): + return + _reasoning_text += reason_delta # Mirror to shared dict so cancel_stream() can persist it (#1361 §A) if stream_id in STREAM_REASONING_TEXT: - STREAM_REASONING_TEXT[stream_id] += str(reason_text) - put('reasoning', {'text': str(reason_text)}) + STREAM_REASONING_TEXT[stream_id] += reason_delta + put('reasoning', {'text': reason_delta}) _metering_reasoning_deltas[0] += 1 meter().record_reasoning(stream_id, _metering_reasoning_deltas[0]) _emit_metering() diff --git a/tests/test_issue_progress_echo_dedupe.py b/tests/test_issue_progress_echo_dedupe.py new file mode 100644 index 00000000..4b0e5109 --- /dev/null +++ b/tests/test_issue_progress_echo_dedupe.py @@ -0,0 +1,177 @@ +import queue +import sys +import types +from typing import Callable, cast +from unittest import mock + + +_MISSING = object() + + +def test_visible_progress_token_reasoning_and_interim_are_deduped(cleanup_test_sessions): + """Progress text can arrive through three Hermes callbacks; WebUI must show it once. + + Some runtimes emit a user-visible progress sentence as a normal token, mirror the + same text through reasoning, and then report it through interim_assistant before + a tool call. The SSE bridge should keep the visible token, suppress the hidden + reasoning echo, and mark interim_assistant as already_streamed so the client and + journal recovery do not append the same paragraph again. + """ + import api.streaming as streaming + + progress = "Gefunden: der Skill-Tab lädt `/api/skill-html?slug=...`." + + class FakeSession: + def __init__(self): + self.session_id = "issue_progress_echo_dedupe" + self.title = "Progress echo" + self.workspace = "/tmp" + self.model = "gpt-test" + self.model_provider = None + self.profile = None + self.personality = None + self.messages = [] + self.context_messages = [] + self.input_tokens = 0 + self.output_tokens = 0 + self.estimated_cost = 0 + self.cache_read_tokens = 0 + self.cache_write_tokens = 0 + self.tool_calls = [] + self.gateway_routing = None + self.gateway_routing_history = [] + self.active_stream_id = "" + self.pending_user_message = None + self.pending_attachments = [] + self.pending_started_at = None + self.context_length = 0 + self.threshold_tokens = 0 + self.last_prompt_tokens = 0 + self.llm_title_generated = True + + def save(self, *args, **kwargs): + pass + + def compact(self): + return { + "session_id": self.session_id, + "title": self.title, + "workspace": self.workspace, + "model": self.model, + "created_at": 0, + "updated_at": 0, + "pinned": False, + "archived": False, + "project_id": None, + "profile": self.profile, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "estimated_cost": self.estimated_cost, + "cache_read_tokens": self.cache_read_tokens, + "cache_write_tokens": self.cache_write_tokens, + "personality": self.personality, + } + + class EchoAgent: + def __init__( + self, + model=None, + provider=None, + base_url=None, + platform=None, + quiet_mode=False, + enabled_toolsets=None, + fallback_model=None, + session_id=None, + session_db=None, + prefill_messages=None, + stream_delta_callback=None, + reasoning_callback=None, + tool_progress_callback=None, + clarify_callback=None, + interim_assistant_callback=None, + **_kwargs, + ): + self.stream_delta_callback = cast(Callable[[str], None], stream_delta_callback) + self.reasoning_callback = cast(Callable[[str], None], reasoning_callback) + self.tool_progress_callback = cast(Callable[..., None], tool_progress_callback) + self.interim_assistant_callback = cast(Callable[[str], None], interim_assistant_callback) + self.context_compressor = None + self.session_prompt_tokens = 0 + self.session_completion_tokens = 0 + self.session_estimated_cost_usd = 0 + self.session_cache_read_tokens = 0 + self.session_cache_write_tokens = 0 + self.reasoning_config = None + self.ephemeral_system_prompt = None + self._last_error = None + + def run_conversation(self, **kwargs): + self.stream_delta_callback(progress) + self.reasoning_callback(progress) + self.tool_progress_callback("reasoning.available", "progress", progress, {}) + self.interim_assistant_callback(progress) + history = kwargs.get("conversation_history", []) + return {"messages": history + [ + {"role": "user", "content": kwargs["persist_user_message"]}, + {"role": "assistant", "content": progress}, + ]} + + def interrupt(self, _message): + pass + + fake_session = FakeSession() + fake_stream_id = "stream_issue_progress_echo_dedupe" + fake_session.active_stream_id = fake_stream_id + fake_queue = queue.Queue() + fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider") + runtime_payload = { + "provider": "openai", + "base_url": None, + "api_mode": "chat_completions", + "command": None, + "args": [], + "credential_pool": None, + } + runtime_payload["api_" + "key"] = "***" + setattr(fake_runtime_module, "resolve_runtime_provider", mock.Mock(return_value=runtime_payload)) + fake_hermes_cli = types.ModuleType("hermes_cli") + setattr(fake_hermes_cli, "runtime_provider", fake_runtime_module) + fake_hermes_state = types.ModuleType("hermes_state") + setattr(fake_hermes_state, "SessionDB", mock.Mock(return_value=None)) + injected = { + "hermes_cli": fake_hermes_cli, + "hermes_cli.runtime_provider": fake_runtime_module, + "hermes_state": fake_hermes_state, + } + saved = {k: sys.modules.get(k, _MISSING) for k in injected} + sys.modules.update(injected) + try: + with mock.patch.object(streaming, "get_session", return_value=fake_session), \ + mock.patch.object(streaming, "_get_ai_agent", return_value=EchoAgent), \ + mock.patch.object(streaming, "resolve_model_provider", return_value=("gpt-test", "openai", None)), \ + mock.patch("api.config.get_config", return_value={}), \ + mock.patch("api.config._resolve_cli_toolsets", return_value=[]): + streaming.STREAMS[fake_stream_id] = fake_queue + streaming._run_agent_streaming( + session_id=fake_session.session_id, + msg_text="scan", + model="gpt-test", + workspace="/tmp", + stream_id=fake_stream_id, + ) + finally: + streaming.STREAMS.pop(fake_stream_id, None) + for k, prev in saved.items(): + if prev is _MISSING: + sys.modules.pop(k, None) + else: + sys.modules[k] = cast(types.ModuleType, prev) + + events = list(fake_queue.queue) + assert [(event, payload) for event, payload in events if event == "token"] == [ + ("token", {"text": progress}) + ] + assert not [payload for event, payload in events if event == "reasoning" and payload.get("text") == progress] + interim = [payload for event, payload in events if event == "interim_assistant"] + assert interim == [{"text": progress, "already_streamed": True}] diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index 1b6230a2..b5e133a5 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -680,9 +680,11 @@ def test_streaming_persists_reasoning_in_session(): assert "_reasoning_text = ''" in src, \ "_reasoning_text variable not initialised in streaming.py" - # on_reasoning must accumulate into _reasoning_text - assert '_reasoning_text += str(text)' in src, \ - "on_reasoning callback does not accumulate into _reasoning_text" + # on_reasoning must accumulate non-echo reasoning into _reasoning_text + assert '_reasoning_text += reasoning_delta' in src, \ + "on_reasoning callback does not accumulate accepted reasoning deltas into _reasoning_text" + assert '_is_visible_output_echo(reasoning_delta)' in src, \ + "on_reasoning callback should suppress reasoning deltas that only echo visible streamed output" # Persistence block must exist before raw_session is built assert "Persist reasoning trace in the session so it survives reload" in src, \