From ce02ea21fbf786e3e4ddf5be5041ad45e9e29621 Mon Sep 17 00:00:00 2001
From: JOJO <1498581755@qq.com>
Date: Fri, 3 Apr 2026 16:09:29 +0800
Subject: [PATCH] fix(stream): prevent tool-call argument leakage into text
 chunks

---
 server/chat_flow_stream_loop.py | 99 +++++++++++++++++++--------------
 1 file changed, 56 insertions(+), 43 deletions(-)

diff --git a/server/chat_flow_stream_loop.py b/server/chat_flow_stream_loop.py
index c3ee766..5820ecf 100644
--- a/server/chat_flow_stream_loop.py
+++ b/server/chat_flow_stream_loop.py
@@ -15,6 +15,7 @@ from .state import get_stop_flag, clear_stop_flag
 
 async def run_streaming_attempts(*, web_terminal, messages, tools, sender, client_sid: str, username: str, conversation_id: Optional[str], current_iteration: int, max_api_retries: int, retry_delay_seconds: int, detected_tool_intent: Dict[str, str], full_response: str, tool_calls: list, current_thinking: str, detected_tools: Dict[str, str], last_usage_payload, in_thinking: bool, thinking_started: bool, thinking_ended: bool, text_started: bool, text_has_content: bool, text_streaming: bool, text_chunk_index: int, last_text_chunk_time, chunk_count: int, reasoning_chunks: int, content_chunks: int, tool_chunks: int, last_finish_reason: Optional[str], accumulated_response: str) -> Dict[str, Any]:
     api_error = None
+    tool_call_stream_active = False
     for api_attempt in range(max_api_retries + 1):
         api_error = None
         if api_attempt > 0:
@@ -36,6 +37,7 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
             content_chunks = 0
             tool_chunks = 0
             last_finish_reason = None
+            tool_call_stream_active = False
 
         # 收集流式响应
         async for chunk in web_terminal.api_client.chat(messages, tools, stream=True):
@@ -128,51 +130,12 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
                 current_thinking += reasoning_content
                 sender('thinking_chunk', {'content': reasoning_content})
 
-            # 处理正常内容
-            if "content" in delta:
-                content = delta["content"]
-                if content:
-                    content_chunks += 1
-                    debug_log(f"  正式内容 #{content_chunks}: {repr(content[:100] if content else 'None')}")
-
-                    if in_thinking and not thinking_ended:
-                        in_thinking = False
-                        thinking_ended = True
-                        sender('thinking_end', {'full_content': current_thinking})
-                        await asyncio.sleep(0.1)
-
-                    if not text_started:
-                        text_started = True
-                        text_streaming = True
-                        sender('text_start', {})
-                        brief_log("模型输出了内容")
-                        await asyncio.sleep(0.05)
-
-                    full_response += content
-                    accumulated_response += content
-                    text_has_content = True
-                    emit_time = time.time()
-                    elapsed = 0.0 if last_text_chunk_time is None else emit_time - last_text_chunk_time
-                    last_text_chunk_time = emit_time
-                    text_chunk_index += 1
-                    log_backend_chunk(
-                        conversation_id,
-                        current_iteration,
-                        text_chunk_index,
-                        elapsed,
-                        len(content),
-                        content[:32]
-                    )
-                    sender('text_chunk', {
-                        'content': content,
-                        'index': text_chunk_index,
-                        'elapsed': elapsed
-                    })
-
             # 收集工具调用 - 实时发送准备状态
-            if "tool_calls" in delta:
+            delta_tool_calls = delta.get("tool_calls")
+            if isinstance(delta_tool_calls, list) and delta_tool_calls:
+                tool_call_stream_active = True
                 tool_chunks += 1
-                for tc in delta["tool_calls"]:
+                for tc in delta_tool_calls:
                     found = False
                     for existing in tool_calls:
                         if existing.get("index") == tc.get("index"):
@@ -261,6 +224,56 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
                                 await asyncio.sleep(0.01)
                         debug_log(f"    新工具: {tool_name}")
 
+            # 处理正常内容
+            if "content" in delta:
+                content = delta["content"]
+                if content:
+                    # 某些供应商在 tool_calls 流式阶段会把参数碎片误放进 content。
+                    # 一旦进入工具调用流，抑制正文 chunk，避免 write_file 参数泄露到前端。
+                    if tool_call_stream_active:
+                        debug_log(
+                            "  抑制可疑正文chunk（tool_call阶段）: "
+                            f"{repr((content or '')[:100])}"
+                        )
+                        continue
+
+                    content_chunks += 1
+                    debug_log(f"  正式内容 #{content_chunks}: {repr(content[:100] if content else 'None')}")
+
+                    if in_thinking and not thinking_ended:
+                        in_thinking = False
+                        thinking_ended = True
+                        sender('thinking_end', {'full_content': current_thinking})
+                        await asyncio.sleep(0.1)
+
+                    if not text_started:
+                        text_started = True
+                        text_streaming = True
+                        sender('text_start', {})
+                        brief_log("模型输出了内容")
+                        await asyncio.sleep(0.05)
+
+                    full_response += content
+                    accumulated_response += content
+                    text_has_content = True
+                    emit_time = time.time()
+                    elapsed = 0.0 if last_text_chunk_time is None else emit_time - last_text_chunk_time
+                    last_text_chunk_time = emit_time
+                    text_chunk_index += 1
+                    log_backend_chunk(
+                        conversation_id,
+                        current_iteration,
+                        text_chunk_index,
+                        elapsed,
+                        len(content),
+                        content[:32]
+                    )
+                    sender('text_chunk', {
+                        'content': content,
+                        'index': text_chunk_index,
+                        'elapsed': elapsed
+                    })
+
         # 检查是否被停止
         client_stop_info = get_stop_flag(client_sid, username)
         if client_stop_info: