From a2d2ddc5a2d09185d06c77b77a1a798543020733 Mon Sep 17 00:00:00 2001 From: Ziheng Zhang <142805986+MuXinCG2004@users.noreply.github.com> Date: Wed, 15 Apr 2026 21:05:47 +0800 Subject: [PATCH] fix(channel): avoid replaying thinking after resume (#154) * fix(channel): avoid replaying thinking after resume * fix(channel): relay fresh thinking after resume * Fix * Fix --------- Co-authored-by: Xi Zhang <106144707+X-iZhang@users.noreply.github.com> --- EvoScientist/channels/consumer.py | 44 +++++---- EvoScientist/stream/display.py | 47 +++++----- tests/test_event_loop.py | 96 +++++++++++++++++++ tests/test_subagent_summarize.py | 147 ++++++++++++++++++++++++++++++ 4 files changed, 294 insertions(+), 40 deletions(-) diff --git a/EvoScientist/channels/consumer.py b/EvoScientist/channels/consumer.py index b20289b..c5418e8 100644 --- a/EvoScientist/channels/consumer.py +++ b/EvoScientist/channels/consumer.py @@ -456,6 +456,8 @@ class InboundConsumer: if channel: await channel.start_typing(msg.chat_id) + _last_sent_thinking: str | None = None + for _hitl_round in range(_MAX_HITL_ROUNDS): final_content = "" thinking_buffer: list[str] = [] @@ -464,6 +466,28 @@ class InboundConsumer: thinking_sent = False interrupt_data: dict | None = None + async def _flush_thinking_buffer( + buffer: list[str] = thinking_buffer, + ) -> bool: + """Send the current thinking buffer, dedup by content.""" + nonlocal thinking_sent, _last_sent_thinking + if not channel or thinking_sent or not buffer: + return False + + full_thinking = "".join(buffer).rstrip() + buffer.clear() + if not full_thinking or full_thinking == _last_sent_thinking: + return False + + await channel.send_thinking_message( + msg.sender_id, + full_thinking, + msg.metadata, + ) + thinking_sent = True + _last_sent_thinking = full_thinking + return True + async for event in _timeout_aiter( stream_agent_events( self.agent, @@ -492,16 +516,7 @@ class InboundConsumer: if event.get("name") == "write_todos" and not todo_sent: todos = event.get("args", {}).get("todos", []) if todos and channel: - if thinking_buffer and not thinking_sent: - full_thinking = "".join(thinking_buffer) - if full_thinking: - await channel.send_thinking_message( - msg.sender_id, - full_thinking, - msg.metadata, - ) - thinking_sent = True - thinking_buffer.clear() + await _flush_thinking_buffer() await channel.send_todo_message( msg.sender_id, _format_todo_list(todos), @@ -533,14 +548,7 @@ class InboundConsumer: break # exit async for to handle ask_user # Flush thinking - if thinking_buffer and not thinking_sent and channel: - full_thinking = "".join(thinking_buffer) - if full_thinking: - await channel.send_thinking_message( - msg.sender_id, - full_thinking, - msg.metadata, - ) + await _flush_thinking_buffer() # No interrupt — normal completion if interrupt_data is None: diff --git a/EvoScientist/stream/display.py b/EvoScientist/stream/display.py index 79c203f..56f47bb 100644 --- a/EvoScientist/stream/display.py +++ b/EvoScientist/stream/display.py @@ -1075,6 +1075,7 @@ def _run_streaming( _state: StreamState | None = None, _hitl_depth: int = 0, _media_sent: set[str] | None = None, + _sent_thinking_text: str | None = None, ) -> str: """Run async streaming and render with Rich Live display. @@ -1087,8 +1088,10 @@ def _run_streaming( show_thinking: Whether to show thinking panel interactive: If True, use simplified final display (no panel) on_thinking: Optional sync callback receiving full thinking text. - Called once when thinking phase ends (transitions to tool/text) - and accumulated thinking >= 200 chars. + Called when thinking ends (transitions to tool/text) and + accumulated thinking >= 200 chars. Uses content-based + deduplication across resume/HITL cycles so the same thinking + is not replayed, but genuinely new thinking is still sent. on_todo: Optional sync callback receiving todo items list. Called once when write_todos tool_call is detected. on_file_write: Optional sync callback receiving the real filesystem path @@ -1100,29 +1103,32 @@ def _run_streaming( The final response text. """ state = _state if _state is not None else StreamState() - _thinking_sent = False _todo_sent = False if _media_sent is None: _media_sent = set() _MIN_THINKING_LEN = 200 async def _consume() -> None: - nonlocal _thinking_sent, _todo_sent + nonlocal _sent_thinking_text, _todo_sent async for event in stream_agent_events( agent, message, thread_id, metadata=metadata ): event_type = state.handle_event(event) - # Send thinking to channel when transitioning away from thinking + # Relay thinking to channel when transitioning away from + # thinking phase. Uses content comparison so that replayed + # thinking after resume is skipped, but genuinely new + # thinking is still delivered. if ( on_thinking - and not _thinking_sent - and state.thinking_text and event_type != "thinking" + and state.thinking_text and len(state.thinking_text) >= _MIN_THINKING_LEN ): - on_thinking(state.thinking_text.rstrip()) - _thinking_sent = True + current = state.thinking_text.rstrip() + if current != _sent_thinking_text: + on_thinking(current) + _sent_thinking_text = current # Send todo list to channel on first write_todos tool_call if ( @@ -1132,15 +1138,6 @@ def _run_streaming( and event.get("name") == "write_todos" and state.todo_items ): - # Flush thinking before todo if not sent yet - if ( - on_thinking - and not _thinking_sent - and state.thinking_text - and len(state.thinking_text) >= _MIN_THINKING_LEN - ): - on_thinking(state.thinking_text.rstrip()) - _thinking_sent = True on_todo(state.todo_items) _todo_sent = True @@ -1310,10 +1307,12 @@ def _run_streaming( loop.run_until_complete(_run_with_refresh()) - # Flush any remaining thinking that wasn't sent during streaming - if on_thinking and not _thinking_sent and state.thinking_text: - if len(state.thinking_text) >= _MIN_THINKING_LEN: - on_thinking(state.thinking_text.rstrip()) + # Flush any remaining thinking that wasn't sent during streaming. + if on_thinking and state.thinking_text: + current = state.thinking_text.rstrip() + if len(current) >= _MIN_THINKING_LEN and current != _sent_thinking_text: + on_thinking(current) + _sent_thinking_text = current # ask_user: check before HITL (ask_user uses the same resume loop) if state.pending_ask_user is not None and _hitl_depth < _MAX_HITL_ITERATIONS: @@ -1324,6 +1323,7 @@ def _run_streaming( from langgraph.types import Command # type: ignore[import-untyped] state.pending_ask_user = None + state.thinking_text = "" # reset accumulation for fresh round return _run_streaming( agent=agent, message=Command(resume=result), @@ -1341,6 +1341,7 @@ def _run_streaming( _state=state, _hitl_depth=_hitl_depth + 1, _media_sent=_media_sent, + _sent_thinking_text=_sent_thinking_text, ) # HITL: check for pending interrupt and handle approval @@ -1353,6 +1354,7 @@ def _run_streaming( from langgraph.types import Command # type: ignore[import-untyped] state.pending_interrupt = None + state.thinking_text = "" # reset accumulation for fresh round return _run_streaming( agent=agent, message=Command(resume={"decisions": decisions}), @@ -1370,6 +1372,7 @@ def _run_streaming( _state=state, _hitl_depth=_hitl_depth + 1, _media_sent=_media_sent, + _sent_thinking_text=_sent_thinking_text, ) elif state.pending_interrupt is not None: _logger.warning( diff --git a/tests/test_event_loop.py b/tests/test_event_loop.py index 88eac24..244f247 100644 --- a/tests/test_event_loop.py +++ b/tests/test_event_loop.py @@ -188,6 +188,102 @@ class TestMultipleStreamingCalls: # Cleanup loop2.close() + def test_recursive_streaming_does_not_resend_same_thinking(self): + """Resumed runs should not replay the original thinking to channels.""" + from EvoScientist.stream.display import _run_streaming + + mock_agent = Mock() + thinking = "Initial plan. " * 20 + stream_calls = 0 + + async def mock_stream(*args, **kwargs): + nonlocal stream_calls + stream_calls += 1 + if stream_calls == 1: + yield {"type": "thinking", "content": thinking} + yield { + "type": "ask_user", + "interrupt_id": "ask-1", + "tool_call_id": "tc-1", + "questions": [{"question": "Continue?"}], + } + return + + yield {"type": "text", "content": "final answer"} + yield {"type": "done", "response": "final answer"} + + sent_thinking: list[str] = [] + + with patch( + "EvoScientist.stream.display.stream_agent_events", + side_effect=mock_stream, + ): + with patch("EvoScientist.stream.display.Live"): + result = _run_streaming( + agent=mock_agent, + message="test message", + thread_id="thread1", + show_thinking=False, + interactive=True, + on_thinking=sent_thinking.append, + ask_user_prompt_fn=lambda _data: { + "answers": ["yes"], + "status": "answered", + }, + ) + + assert result == "final answer" + assert sent_thinking == [thinking.rstrip()] + + def test_recursive_streaming_sends_new_thinking_after_resume(self): + """Genuinely new thinking in resumed rounds should be relayed.""" + from EvoScientist.stream.display import _run_streaming + + mock_agent = Mock() + thinking_r1 = "Initial plan. " * 20 + thinking_r2 = "Revised plan. " * 20 + stream_calls = 0 + + async def mock_stream(*args, **kwargs): + nonlocal stream_calls + stream_calls += 1 + if stream_calls == 1: + yield {"type": "thinking", "content": thinking_r1} + yield { + "type": "ask_user", + "interrupt_id": "ask-1", + "tool_call_id": "tc-1", + "questions": [{"question": "Continue?"}], + } + return + + yield {"type": "thinking", "content": thinking_r2} + yield {"type": "text", "content": "final answer"} + yield {"type": "done", "response": "final answer"} + + sent_thinking: list[str] = [] + + with patch( + "EvoScientist.stream.display.stream_agent_events", + side_effect=mock_stream, + ): + with patch("EvoScientist.stream.display.Live"): + result = _run_streaming( + agent=mock_agent, + message="test message", + thread_id="thread1", + show_thinking=False, + interactive=True, + on_thinking=sent_thinking.append, + ask_user_prompt_fn=lambda _data: { + "answers": ["yes"], + "status": "answered", + }, + ) + + assert result == "final answer" + assert sent_thinking == [thinking_r1.rstrip(), thinking_r2.rstrip()] + class TestEventLoopThreadSafety: """Tests for thread safety edge cases.""" diff --git a/tests/test_subagent_summarize.py b/tests/test_subagent_summarize.py index 259a2d3..bd8891c 100644 --- a/tests/test_subagent_summarize.py +++ b/tests/test_subagent_summarize.py @@ -360,6 +360,153 @@ class TestConsumerSubagentTextFallback: _run(_test()) + def test_duplicate_thinking_not_relayed_across_resume_rounds(self): + """Repeated thinking from resumed rounds should only be sent once.""" + bus = MessageBus() + mgr = ChannelManager(bus) + mgr.register(_StubChannel()) + + channel = mgr.get_channel("stub") + assert channel is not None + channel.send_thinking_message = AsyncMock() + + consumer = InboundConsumer( + bus=bus, + manager=mgr, + agent=MagicMock(), + thread_id="", + max_concurrent=2, + max_pending=10, + inference_timeout=5.0, + drain_timeout=1.0, + send_thinking=True, + ) + consumer._resolve_ask_user = AsyncMock( # type: ignore[method-assign] + return_value={"answers": ["yes"], "status": "answered"} + ) + + thinking = "Initial plan. " * 20 + stream_calls = 0 + + async def _fake_stream(agent, message, thread_id, **kwargs): + nonlocal stream_calls + stream_calls += 1 + if stream_calls == 1: + yield {"type": "thinking", "content": thinking} + yield { + "type": "ask_user", + "interrupt_id": "ask-1", + "tool_call_id": "tc-1", + "questions": [{"question": "Continue?"}], + } + return + + yield {"type": "thinking", "content": thinking} + yield {"type": "text", "content": "final answer"} + yield {"type": "done", "content": "final answer"} + + async def _test(): + with patch( + "EvoScientist.stream.events.stream_agent_events", + new=_fake_stream, + ): + await bus.publish_inbound( + BusInbound( + channel="stub", + sender_id="u1", + chat_id="c1", + content="analyze papers", + ) + ) + + task = asyncio.create_task(consumer.run()) + outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=5.0) + + assert outbound.content == "final answer" + assert channel.send_thinking_message.await_count == 1 + call = channel.send_thinking_message.await_args_list[0] + assert call.args[1] == thinking.rstrip() + + await consumer.stop() + await task + + _run(_test()) + + def test_new_thinking_relayed_after_resume(self): + """Genuinely different thinking in round 2 should be sent.""" + bus = MessageBus() + mgr = ChannelManager(bus) + mgr.register(_StubChannel()) + + channel = mgr.get_channel("stub") + assert channel is not None + channel.send_thinking_message = AsyncMock() + + consumer = InboundConsumer( + bus=bus, + manager=mgr, + agent=MagicMock(), + thread_id="", + max_concurrent=2, + max_pending=10, + inference_timeout=5.0, + drain_timeout=1.0, + send_thinking=True, + ) + consumer._resolve_ask_user = AsyncMock( # type: ignore[method-assign] + return_value={"answers": ["yes"], "status": "answered"} + ) + + thinking_r1 = "Initial plan. " * 20 + thinking_r2 = "Revised plan. " * 20 + stream_calls = 0 + + async def _fake_stream(agent, message, thread_id, **kwargs): + nonlocal stream_calls + stream_calls += 1 + if stream_calls == 1: + yield {"type": "thinking", "content": thinking_r1} + yield { + "type": "ask_user", + "interrupt_id": "ask-1", + "tool_call_id": "tc-1", + "questions": [{"question": "Continue?"}], + } + return + + yield {"type": "thinking", "content": thinking_r2} + yield {"type": "text", "content": "final answer"} + yield {"type": "done", "content": "final answer"} + + async def _test(): + with patch( + "EvoScientist.stream.events.stream_agent_events", + new=_fake_stream, + ): + await bus.publish_inbound( + BusInbound( + channel="stub", + sender_id="u1", + chat_id="c1", + content="analyze papers", + ) + ) + + task = asyncio.create_task(consumer.run()) + outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=5.0) + + assert outbound.content == "final answer" + assert channel.send_thinking_message.await_count == 2 + call1 = channel.send_thinking_message.await_args_list[0] + call2 = channel.send_thinking_message.await_args_list[1] + assert call1.args[1] == thinking_r1.rstrip() + assert call2.args[1] == thinking_r2.rstrip() + + await consumer.stop() + await task + + _run(_test()) + def test_no_response_fallback_when_both_empty(self): """When both final_content and subagent_text are empty, 'No response' is used.""" events = [