feat(stream): 流式断流「清除重来」自动重试机制

后端↔API 流式连接在输出中途断开时，旧行为直接终止任务。现改为：
- 网络类断流（status_code 为空）有接收也重试，与零接收共用 5 次
  额度、间隔 10s；4xx 业务错误保持原逻辑
- 重试前发 stream_reset 事件，前端清除本轮已渲染的半截内容
  （思考块含已闭合的/文本/preparing 工具条目），显示「连接中断，
  正在重试（第 n/5 次）…」占位，socket 与 REST 轮询双通道生效
- 修复断流作废轮 usage 重复计数
- 子智能体同构改造：网络断流重试耗尽转 idle 汇报 Team Leader，
  重试前将已推送的「正在调用」工具条目标记 cancelled

modules/sub_agent/task.py

@@ -37,14 +37,19 @@ class SubAgentModelCallError(RuntimeError):
     """子智能体模型请求失败。
 
     received_any=False：请求阶段失败（未收到任何文本/思考/工具调用），可重试；
-    received_any=True：已开始收到流内容后断开（输出期间断开），直接失败不重试。
-    语义与主智能体 run_streaming_attempts 的 can_retry 判定一致
-    （not full_response and not tool_calls and not current_thinking）。
+    received_any=True：已开始收到流内容后断开（输出期间断开）——
+    网络类断流（is_network_error=True）同样可重试（清除重来，与主智能体
+    run_streaming_attempts 的放宽判定一致）；非网络错误（4xx 等业务错误）
+    有接收时不重试。
+    calling_tools：断流时流中已推送「正在调用」进度事件的工具列表，
+    重试前由 _run_loop 标记取消，避免前端进度弹窗永久卡在「调用中」。
     """
 
-    def __init__(self, message: str, *, received_any: bool = False):
+    def __init__(self, message: str, *, received_any: bool = False, is_network_error: bool = True, calling_tools: Optional[List[Dict[str, Any]]] = None):
         super().__init__(message)
         self.received_any = received_any
+        self.is_network_error = is_network_error
+        self.calling_tools = calling_tools or []
 
 
 # 模型请求失败重试策略（与主智能体 max_api_retries=4 / retry_delay_seconds=10 对齐：
@@ -331,9 +336,10 @@ class SubAgentTask:
             # 安全点：写入延迟的上下文通知（不触发新一轮工作，仅让下一轮模型调用看到）
             self._flush_pending_notifications()
 
-            # 模型请求（带失败重试，与主智能体同构：最多 5 次尝试、重试间隔 10s、
-            # 仅当零接收——未收到任何文本/思考/工具调用——时重试；
-            # 已开始收到内容后断开属于输出期间故障，直接失败不重试）
+            # 模型请求（带失败重试，与主智能体同构：最多 5 次尝试、重试间隔 10s）。
+            # 重试范围：零接收一律可重试；有接收（输出中途断开）仅网络类断流可重试，
+            # 采取「清除重来」——半截内容只在 _call_model 局部变量与已推送的 calling
+            # 进度事件里，转发给主智能体的输出只发生在流完整结束后，无副作用。
             assistant_message = ""
             reasoning = ""
             tool_calls = []
@@ -348,7 +354,7 @@ class SubAgentTask:
                     break
                 except SubAgentModelCallError as exc:
                     call_error = exc
-                    can_retry = api_attempt < _SUB_AGENT_MAX_API_RETRIES and not exc.received_any
+                    can_retry = api_attempt < _SUB_AGENT_MAX_API_RETRIES and (exc.is_network_error or not exc.received_any)
                     ma_debug(
                         "sub_agent_api_attempt_failed",
                         task_id=self.task_id,
@@ -363,6 +369,17 @@ class SubAgentTask:
                     )
                     if not can_retry:
                         break
+                    # 有接收断流重试（清除重来）：本轮已推送的「正在调用」工具进度
+                    # 条目标记取消，避免前端进度弹窗永久卡在「调用中」；
+                    # 重试成功后模型若再次调用会推送新的条目（新一轮工具 id 不同）。
+                    for calling_tool in exc.calling_tools:
+                        self.emit("progress", {
+                            "id": calling_tool.get("id"),
+                            "tool": calling_tool.get("name", ""),
+                            "status": "cancelled",
+                            "args": {},
+                            "ts": int(time.time() * 1000),
+                        })
                     # 重试本身也是一次额外 API 请求
                     self.stats["api_calls"] += 1
                     # 重试等待期间必须响应软停止/硬取消
@@ -376,8 +393,8 @@ class SubAgentTask:
                     continue
 
             if call_error is not None:
-                if self.multi_agent_mode and not call_error.received_any:
-                    # 5 次尝试全部失败（均为零接收）：子智能体不终止，转为 idle
+                if self.multi_agent_mode and (call_error.is_network_error or not call_error.received_any):
+                    # 5 次尝试全部失败（零接收或网络断流）：子智能体不终止，转为 idle
                     # 并向 Team Leader 汇报错误，等待检查网络后重新下达指令
                     error_report = tr(
                         "sub_agent_task.model_call_failed_idle",
@@ -396,8 +413,8 @@ class SubAgentTask:
                     self._idle = True
                     self._persist_conversation(partial_summary=error_report[:200])
                     continue
-                # 输出期间断开（已开始收到内容）或传统后台模式：直接失败，
-                # 异常上抛由 run() 捕获后 _write_failure 落盘 failed 状态
+                # 非网络业务错误（有接收，理论罕见）或传统后台模式（重试已耗尽）：
+                # 直接失败，异常上抛由 run() 捕获后 _write_failure 落盘 failed 状态
                 if self.multi_agent_mode:
                     # 同步把失败状态告知 Team Leader，避免主智能体无感知空等
                     self._forward_output_to_master(
@@ -779,15 +796,30 @@ class SubAgentTask:
                     raise asyncio.CancelledError()
                 if chunk.get("error"):
                     error_info = chunk.get("error")
+                    is_network = True
                     if isinstance(error_info, dict):
                         error_text = (
                             error_info.get("error_message")
                             or error_info.get("error_text")
                             or str(error_info)
                         )
+                        # status_code 为空 = 网络类断流（连接错误/超时/远端断开）；
+                        # 有值（4xx 等 HTTP 业务错误）不参与有接收重试放宽
+                        is_network = error_info.get("status_code") is None
                     else:
                         error_text = str(error_info)
-                    raise SubAgentModelCallError(tr("sub_agent_task2.api_call_failed", error=error_text), received_any=received_any)
+                    # 收集本轮已推送「正在调用」进度的工具，重试前需标记取消
+                    calling_tools = [
+                        {"id": tc.get("id"), "name": tc.get("function", {}).get("name", "")}
+                        for tc in tool_calls
+                        if tc.get("id") and tc.get("function", {}).get("name")
+                    ]
+                    raise SubAgentModelCallError(
+                        tr("sub_agent_task2.api_call_failed", error=error_text),
+                        received_any=received_any,
+                        is_network_error=is_network,
+                        calling_tools=calling_tools,
+                    )
                 choice = (chunk.get("choices") or [{}])[0]
                 delta = choice.get("delta") or {}
                 if delta.get("content"):
@@ -844,7 +876,17 @@ class SubAgentTask:
             raise
         except Exception as exc:
             # chat 流本身抛出的漏网异常（如底层连接错误未被转为 error chunk 时）
-            raise SubAgentModelCallError(tr("sub_agent_task2.api_call_exception", error=exc), received_any=received_any) from exc
+            calling_tools = [
+                {"id": tc.get("id"), "name": tc.get("function", {}).get("name", "")}
+                for tc in tool_calls
+                if tc.get("id") and tc.get("function", {}).get("name")
+            ]
+            raise SubAgentModelCallError(
+                tr("sub_agent_task2.api_call_exception", error=exc),
+                received_any=received_any,
+                is_network_error=True,
+                calling_tools=calling_tools,
+            ) from exc
 
         return assistant_message, reasoning, tool_calls, usage
 

server/chat_flow_stream_loop.py

@@ -319,7 +319,9 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
                 }
 
         # === API响应完成后只计算输出token ===
-        if last_usage_payload:
+        # 断流作废轮不计 usage：本论响应不完整，重试成功后会按新一轮 usage 计数，
+        # 若在 error 分支判断之前 apply 会导致同一回答重复计数。
+        if last_usage_payload and not api_error:
             try:
                 web_terminal.context_manager.apply_usage_statistics(last_usage_payload)
                 debug_log(
@@ -367,12 +369,13 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
                     error_message = f"API 请求失败（HTTP {error_status}）"
                 else:
                     error_message = "API 请求失败"
-            can_retry = (
-                api_attempt < max_api_retries
-                and not full_response
-                and not tool_calls
-                and not current_thinking
-            )
+            # 重试判定：网络类断流（status_code 为空：连接错误/超时/远端断开，
+            # 区别于 HTTP 业务错误）允许「有接收」重试——半截内容只存在于局部变量
+            # 与前端，后端历史与磁盘均未写入，清除重来无副作用；HTTP 业务错误
+            # （4xx 等，实际上不可能有接收）保持原逻辑仅零接收重试。
+            has_partial_output = bool(full_response or tool_calls or current_thinking)
+            is_network_error = error_status is None
+            can_retry = api_attempt < max_api_retries and (is_network_error or not has_partial_output)
             sender('error', {
                 'message': error_message,
                 'status_code': error_status,
@@ -388,6 +391,17 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien
                 'max_attempts': max_api_retries + 1
             })
             if can_retry:
+                if has_partial_output:
+                    # 「清除重来」：通知前端清掉本轮 attempt 已渲染的半截内容
+                    # （思考块含已闭合的 / 文本 / preparing 工具条目），重试的新内容
+                    # 将由新一轮 thinking_start/text_start/tool_preparing 重新推送。
+                    # 后端内存历史与磁盘均无半截状态，无需清理。
+                    sender('stream_reset', {
+                        'reason': 'stream_disconnected',
+                        'attempt': api_attempt + 2,
+                        'max_attempts': max_api_retries + 1,
+                        'retry_in': retry_delay_seconds,
+                    })
                 try:
                     profile = get_model_profile(getattr(web_terminal, "model_key", None))
                     web_terminal.apply_model_profile(profile)

static/src/app.ts

@@ -93,7 +93,8 @@ const appOptions = {
       chatCompleteTextAction: 'completeText',
       chatAddSystemMessage: 'addSystemMessage',
       chatEnsureAssistantMessage: 'ensureAssistantMessage',
-      chatClearStreamingResidualState: 'clearStreamingResidualState'
+      chatClearStreamingResidualState: 'clearStreamingResidualState',
+      chatResetStreamingAttemptActions: 'resetStreamingAttemptActions'
     }),
     ...mapActions(useInputStore, {
       inputSetFocused: 'setInputFocused',

static/src/app/methods/taskPolling/aiStream.ts

@@ -260,5 +260,27 @@ export const aiStreamMethods = {
     this.chatCompleteTextAction(full);
     this.$forceUpdate();
     this.monitorEndModelOutput();
+  },
+  handleStreamReset(data: any) {
+    // 断流重试「清除重来」：移除本轮 attempt 已渲染的半截内容
+    // （思考块含已闭合的 / 文本 / preparing 工具条目），重试内容将重新推送。
+    debugLog('[TaskPolling] 断流重试，清理本轮半截内容');
+    const retryAttempt = Number(data?.attempt) || 0;
+    const retryMax = Number(data?.max_attempts) || 0;
+    const retryLabel = retryAttempt && retryMax
+      ? t('appTasks.streamRetrying', { attempt: retryAttempt, max: retryMax })
+      : t('appTasks.streamRetryingGeneric');
+    const removedActions = this.chatResetStreamingAttemptActions?.(retryLabel) || [];
+    for (const action of removedActions) {
+      if (action?.id) {
+        this.preparingTools?.delete(action.id);
+      }
+      if (action?.type === 'tool') {
+        this.toolUnregisterAction?.(action);
+      }
+    }
+    this.chatClearThinkingLocks?.();
+    this.monitorEndModelOutput();
+    this.$forceUpdate();
   }
 };

static/src/app/methods/taskPolling/lifecycle.ts

@@ -230,6 +230,12 @@ export const lifecycleMethods = {
         this.handleTextEnd(eventData, eventIdx);
         break;
 
+      case 'stream_reset':
+        // 断流重试「清除重来」：清理本轮 attempt 半截内容；
+        // 历史重放时同样安全（后续事件会把新一轮内容重新渲染出来）
+        this.handleStreamReset(eventData, eventIdx);
+        break;
+
       case 'tool_preparing':
         this.handleToolPreparing(eventData, eventIdx);
         break;

static/src/composables/useLegacySocket.ts

@@ -1142,6 +1142,41 @@ export async function initializeLegacySocket(ctx: any) {
       ctx.apiRequestPending = true;
     });
 
+    // 断流重试「清除重来」：后端与 API 的流式连接在输出中途断开、即将自动重试，
+    // 清除本轮 attempt 已渲染的半截内容（思考块含已闭合的 / 文本 / preparing 工具条目），
+    // 重试的新内容会由新一轮 thinking_start/text_start/tool_preparing 重新推送。
+    ctx.socket.on('stream_reset', (data) => {
+      if (data?.conversation_id && data.conversation_id !== ctx.currentConversationId) {
+        return;
+      }
+      // 轮询模式下由任务事件流重放处理（lifecycle.ts case 'stream_reset'）
+      if (ctx.usePollingMode && !ctx.waitingForSubAgent) {
+        return;
+      }
+      socketLog('断流重试，清理本轮半截内容', data);
+      // 清逐字流缓冲，防止半截缓冲在重置后继续渲染
+      resetStreamingBuffer();
+      const retryAttempt = Number(data?.attempt) || 0;
+      const retryMax = Number(data?.max_attempts) || 0;
+      const retryLabel = retryAttempt && retryMax
+        ? t('appTasks.streamRetrying', { attempt: retryAttempt, max: retryMax })
+        : t('appTasks.streamRetryingGeneric');
+      const removedActions = ctx.chatResetStreamingAttemptActions?.(retryLabel) || [];
+      for (const action of removedActions) {
+        if (action?.id) {
+          ctx.preparingTools?.delete(action.id);
+        }
+        if (action?.type === 'tool') {
+          ctx.toolUnregisterAction?.(action);
+        }
+      }
+      if (typeof ctx.chatClearThinkingLocks === 'function') {
+        ctx.chatClearThinkingLocks();
+      }
+      ctx.monitorEndModelOutput?.();
+      ctx.$forceUpdate();
+    });
+
     // 思考流开始
     ctx.socket.on('thinking_start', (data) => {
       // 对话定向事件：只响应当前对话，避免运行中对话的流式输出渲染到 /new 等空白页。

static/src/composables/useMarkdownRenderer.ts

@@ -344,9 +344,11 @@ function transformMathBlocks(raw: string): string {
     return `<span class="math-block" data-latex="${escapeHtmlAttribute(trimmed)}" data-display="block"></span>`;
   });
 
-  // 行内 $...$（排除 $ 本身和反斜杠转义）
+  // 行内 $...$（Pandoc 规则：开 $ 后非空白、闭 $ 前非空白、闭 $ 后不跟数字，
+  // 避免把 "$10 ≈ ¥72 | $5" 这类相邻货币 $ 误配对成公式——配对区间若跨过
+  // 表格 | 分隔符，生成的占位 span 会被表格解析拦腰切断，显示为原始文本）
   protectedText = protectedText.replace(
-    /(?<![\\$])\$([^$\n]+?)\$(?![\\$])/g,
+    /(?<![\\$])\$(?!\s)([^$\n]+?)(?<!\s)\$(?![\\$\d])/g,
     (match, latex: string) => {
       const trimmed = latex.replace(/\s+/g, ' ').trim();
       if (!trimmed) return match;

static/src/locales/en-US/appTasks.ts

@@ -64,6 +64,8 @@ export default {
   // ── Task errors & retry (lifecycle.ts) ──
   retrySoonTitle: 'Retrying soon',
   retryInSeconds: 'Retrying in {n}s (attempt {attempt}/{max})\nError: {error}',
+  streamRetrying: 'Connection interrupted, retrying (attempt {attempt}/{max})…',
+  streamRetryingGeneric: 'Connection interrupted, retrying…',
   toolCallFailed: 'Tool call failed',
   taskFailedTitle: 'Task failed',
   apiErrorTitle: 'API call failed',

static/src/locales/zh-CN/appTasks.ts

@@ -65,6 +65,8 @@ export default {
   // ── 任务错误与重试（lifecycle.ts） ──
   retrySoonTitle: '即将重试',
   retryInSeconds: '将在 {n} 秒后重试（第 {attempt}/{max} 次）\n错误：{error}',
+  streamRetrying: '连接中断，正在重试（第 {attempt}/{max} 次）…',
+  streamRetryingGeneric: '连接中断，正在重试…',
   toolCallFailed: '工具调用失败',
   taskFailedTitle: '任务执行失败',
   apiErrorTitle: 'API 调用失败',

static/src/stores/chat.ts

@@ -419,6 +419,51 @@ export const useChatStore = defineStore('chat', {
         }
       }
     },
+    // 断流重试「清除重来」：移除当前 assistant 消息中本轮 API 尝试已渲染的
+    // 半截内容——从 actions 末尾往前删除思考（含已闭合）/文本/preparing 工具条目，
+    // 遇到上一轮迭代的已完成工具结果等非本轮内容即停。复位流式标志；若消息
+    // 因此被清空则恢复「生成中」占位（label 通常为重试提示文案）。
+    // 返回被移除的 action 对象列表，供调用方清理 preparingTools / 工具注册表。
+    resetStreamingAttemptActions(label: string = ''): any[] {
+      const removed: any[] = [];
+      let msg: any = null;
+      if (
+        this.currentMessageIndex >= 0 &&
+        this.messages[this.currentMessageIndex]?.role === 'assistant'
+      ) {
+        msg = this.messages[this.currentMessageIndex];
+      } else {
+        for (let i = this.messages.length - 1; i >= 0; i--) {
+          if (this.messages[i]?.role === 'assistant') {
+            msg = this.messages[i];
+            break;
+          }
+        }
+      }
+      if (!msg || !Array.isArray(msg.actions)) return removed;
+      while (msg.actions.length) {
+        const last = msg.actions[msg.actions.length - 1];
+        const isThinking = last?.type === 'thinking';
+        const isText = last?.type === 'text';
+        const isPreparingTool =
+          last?.type === 'tool' && String(last?.tool?.status || '').toLowerCase() === 'preparing';
+        if (!isThinking && !isText && !isPreparingTool) break;
+        removed.push(last);
+        msg.actions.pop();
+      }
+      msg.currentStreamingType = null;
+      msg.activeThinkingId = null;
+      msg.streamingThinking = '';
+      msg.streamingText = '';
+      if (msg.actions.length === 0) {
+        msg.awaitingFirstContent = true;
+        msg.generatingLabel = label;
+      } else {
+        msg.awaitingFirstContent = false;
+        msg.generatingLabel = '';
+      }
+      return removed;
+    },
     addSystemMessage(content: string, meta: any = null) {
       // 与历史重建保持一致：子智能体/后台完成通知作为独立 assistant 消息渲染，
       // 避免运行时与刷新后在垂直间距上不一致。