fix(multi-agent): 跨事件循环问答、线程安全取消、任务结束汇报格式、工具参数可选

core/main_terminal_parts/tools_execution.py

@@ -2,6 +2,7 @@ import asyncio
 import json
 import os
 import time
+import uuid
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set
@@ -1755,16 +1756,13 @@ class MainTerminalToolsExecutionMixin:
                                         # 构造 task_message（作为 Team Leader 的任务发布）
                                         from modules.multi_agent.state import build_master_dispatch_text
                                         task_message = build_master_dispatch_text(arguments.get("task", ""))
-                                        # 加载并覆盖作业：multi_agent_mode/sub_agent_manager.create_sub_agent 的作业路径
                                         summary_text = (arguments.get("summary") or f"{role.name}作业")[:80]
-                                        deliverables_dir = arguments.get("deliverables_dir", f"sub_agent_results/agent_{agent_id}")
                                         thinking_mode = arguments.get("thinking_mode") or role.thinking_mode or "fast"
                                         # 走原行 发事件创建（避免后期重建提供重复工能重费，直接使用 multi_agent_mode=True 调用）
                                         result = self.sub_agent_manager.create_sub_agent(
                                             agent_id=int(agent_id),
                                             summary=summary_text,
                                             task=arguments.get("task", ""),
-                                            deliverables_dir=deliverables_dir,
                                             run_in_background=bool(arguments.get("run_in_background", True)),
                                             timeout_seconds=arguments.get("timeout_seconds"),
                                             conversation_id=conv_id,
@@ -1773,6 +1771,8 @@ class MainTerminalToolsExecutionMixin:
                                             multi_agent_mode=True,
                                             role_id=role_id,
                                             display_name=display_name,
+                                            system_prompt=system_prompt,
+                                            task_message=task_message,
                                         )
                                         # 在多智能体模式下，主进程 create_sub_agent 总是后台启动，
                                         # 主智能体不需要阻塞等待，而是通过子智能体输出转发拿进度。
@@ -1887,11 +1887,13 @@ class MainTerminalToolsExecutionMixin:
                                 if not state:
                                     result = {"success": False, "error": "多智能体状态未就绪"}
                                 else:
-                                    # 构造提问并插入子对话（子智能体下一轮 assistant 输出作为回答插入主对话）
+                                    question_id = f"ask_sub_agent_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"
+                                    # 构造提问并插入子对话（子智能体下一轮 assistant 输出作为回答返回到工具结果）
                                     text = format_multi_agent_message(
                                         display_name="Team Leader",
                                         msg_type=TYPE_ASK,
                                         content=question,
+                                        msg_id=question_id,
                                         target=state.get_instance(agent_id).display_name if state.get_instance(agent_id) else f"Agent_{agent_id}",
                                     )
                                     ok = self.sub_agent_manager.inject_message_to_sub_agent(agent_id, text)
@@ -1900,7 +1902,7 @@ class MainTerminalToolsExecutionMixin:
                                     else:
                                         # 阻塞等待子智能体下一轮输出作为回答
                                         answer = await state.wait_for_answer(
-                                            question_id=f"ask_sub_agent_{int(time.time())}",
+                                            question_id=question_id,
                                             agent_id=agent_id,
                                             timeout=timeout,
                                         )

modules/multi_agent/state.py

@@ -20,6 +20,7 @@ from __future__ import annotations
 import asyncio
 import json
 import uuid
+from asyncio import AbstractEventLoop
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -48,6 +49,7 @@ def format_multi_agent_message(
     msg_id: Optional[str] = None,
     target: Optional[str] = None,
     extra_attrs: Optional[Dict[str, str]] = None,
+    msg_type_text: Optional[str] = None,
 ) -> str:
     """按统一格式构造 user 消息字符串。
 
@@ -58,15 +60,17 @@ def format_multi_agent_message(
         msg_id: 消息 id；不传则自动生成
         target: 接收方显示名（用于子→子 提问时标明对谁提问）
         extra_attrs: 额外标签属性（如 question_id="ask_xxx"）
+        msg_type_text: 覆盖默认的中文消息类型文案（如"任务结束汇报"）
     """
     if not msg_id:
         msg_id = f"msg_{uuid.uuid4().hex[:10]}"
 
+    type_label = msg_type_text or msg_type_to_text(msg_type)
     # 第一行：自然语言前缀（含 target 标识）
     if target:
-        prefix = f"来自 {display_name} 向 {target} 的{msg_type_to_text(msg_type)}"
+        prefix = f"来自 {display_name} 向 {target} 的{type_label}"
     else:
-        prefix = f"来自 {display_name} 的{msg_type_to_text(msg_type)}"
+        prefix = f"来自 {display_name} 的{type_label}"
 
     # 第二行：id
     id_line = f"id: {msg_id}"
@@ -114,13 +118,14 @@ def build_master_dispatch_text(task: str, msg_id: Optional[str] = None) -> str:
     )
 
 
-def build_sub_agent_output_text(display_name: str, content: str, msg_id: Optional[str] = None) -> str:
+def build_sub_agent_output_text(display_name: str, content: str, msg_id: Optional[str] = None, *, is_final: bool = False) -> str:
     """子智能体输出（进度或完成）插入到主对话的 user 消息文本。"""
     return format_multi_agent_message(
         display_name=display_name,
         msg_type=TYPE_OUTPUT,
         content=content,
         msg_id=msg_id,
+        msg_type_text="任务结束汇报" if is_final else "任务进度输出",
     )
 
 
@@ -223,6 +228,10 @@ class MultiAgentState:
         # ask_master / ask_other_agent 的等待 future
         # key = question_id, value = asyncio.Future (结果为 answer str 或 Exception)
         self.pending_questions: Dict[str, asyncio.Future] = {}
+        # question_id -> 创建 future 时所在的事件循环，用于跨循环安全 set_result
+        self.pending_question_loops: Dict[str, AbstractEventLoop] = {}
+        # 回答早于 wait_for_answer 注册时先暂存
+        self.pending_answers: Dict[str, str] = {}
         # 一个 agent 可能同时只阻塞在一个 ask 工具上（最简实现）
         # key = agent_id, value = question_id（表示当前 agent 正阻塞等待）
         self.agent_blocking_question: Dict[int, str] = {}
@@ -285,25 +294,26 @@ class MultiAgentState:
 
         返回 answer 字符串；超时/取消抛 asyncio.TimeoutError 或 CancelledError。
         """
+        # 如果回答已经提前到达，直接返回
+        if question_id in self.pending_answers:
+            return self.pending_answers.pop(question_id)
         if question_id in self.pending_questions:
             raise RuntimeError(f"question_id 已存在: {question_id}")
-        fut: asyncio.Future = asyncio.get_event_loop().create_future()
+        loop = asyncio.get_running_loop()
+        fut: asyncio.Future = loop.create_future()
         self.pending_questions[question_id] = fut
+        self.pending_question_loops[question_id] = loop
         self.agent_blocking_question[agent_id] = question_id
         try:
             return await asyncio.wait_for(fut, timeout=timeout)
-        except asyncio.TimeoutError:
-            raise
         finally:
             self.pending_questions.pop(question_id, None)
+            self.pending_question_loops.pop(question_id, None)
             if self.agent_blocking_question.get(agent_id) == question_id:
                 self.agent_blocking_question.pop(agent_id, None)
 
-    def provide_answer(self, question_id: str, answer: str) -> bool:
+    async def _do_provide_answer(self, question_id: str, answer: str) -> bool:
-        """主/其他子智能体 answer_* 工具调用时回写答案。
+        """在同 future 所属事件循环内设置结果。"""
-
-        返回 True 表示找到等待中的 future；False 表示无等待方或已超时。
-        """
         fut = self.pending_questions.get(question_id)
         if not fut or fut.done():
             return False
@@ -313,6 +323,35 @@ class MultiAgentState:
             return False
         return True
 
+    def provide_answer(self, question_id: str, answer: str) -> bool:
+        """主/其他子智能体 answer_* 工具调用时回写答案。
+
+        返回 True 表示找到等待中的 future；False 表示无等待方或已超时。
+        支持跨事件循环调用（例如主对话循环回答子智能体循环里的提问）。
+        """
+        # 如果 wait_for_answer 还没注册，先把答案暂存
+        if question_id not in self.pending_questions:
+            self.pending_answers[question_id] = answer
+            return True
+        fut = self.pending_questions.get(question_id)
+        if not fut or fut.done():
+            self.pending_answers[question_id] = answer
+            return False
+        loop = self.pending_question_loops.get(question_id)
+        if loop is None:
+            try:
+                loop = fut.get_loop()
+            except Exception:
+                pass
+        if loop is not None:
+            try:
+                asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), loop)
+                return True
+            except Exception:
+                pass
+        # 同循环回退
+        return asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), asyncio.get_event_loop()).result(timeout=5)
+
     def is_agent_blocking(self, agent_id: int) -> bool:
         return agent_id in self.agent_blocking_question
 

modules/multi_agent/tools.py

@@ -53,10 +53,6 @@ def _master_tool_create_sub_agent() -> Dict[str, Any]:
                         "type": "integer",
                         "description": "（可选）手动指定实例编号；不传时自动递增。",
                     },
-                    "deliverables_dir": {
-                        "type": "string",
-                        "description": "（可选）交付目录相对路径，留空则用 sub_agent_results/agent_{N}。",
-                    },
                     "timeout_seconds": {"type": "integer", "description": "超时秒数，默认 600。"},
                     "thinking_mode": {
                         "type": "string",
@@ -64,7 +60,7 @@ def _master_tool_create_sub_agent() -> Dict[str, Any]:
                         "description": "（可选）覆盖角色默认思考模式。不填使用角色配置。",
                     },
                 }),
-                "required": ["role_id", "task", "thinking_mode"],
+                "required": ["role_id", "task"],
             },
         },
     }

modules/sub_agent/creation.py

@@ -48,7 +48,7 @@ class SubAgentCreationMixin:
         if agent_id not in used:
             used.append(agent_id)
 
-    def _validate_create_params(self, agent_id: Optional[int], summary: str, task: str, target_dir: str) -> Optional[str]:
+    def _validate_create_params(self, agent_id: Optional[int], summary: str, task: str, target_dir: Optional[str], *, multi_agent_mode: bool = False) -> Optional[str]:
         if agent_id is None:
             return "子智能体代号不能为空"
         try:
@@ -61,7 +61,8 @@ class SubAgentCreationMixin:
             return "任务摘要不能为空"
         if not task or not task.strip():
             return "任务详情不能为空"
-        if target_dir is None:
+        # 多智能体模式不需要交付目录
+        if not multi_agent_mode and target_dir is None:
             return "指定文件夹不能为空"
         return None
 
@@ -69,8 +70,11 @@ class SubAgentCreationMixin:
         suffix = uuid.uuid4().hex[:6]
         return f"sub_{agent_id}_{int(time.time())}_{suffix}"
 
-    def _resolve_deliverables_dir(self, relative_dir: str) -> Path:
+    def _resolve_deliverables_dir(self, relative_dir: Optional[str], *, multi_agent_mode: bool = False) -> Path:
-        relative_dir = relative_dir.strip() if relative_dir else ""
+        relative_dir = (relative_dir or "").strip()
+        # 多智能体模式：没有交付目录概念，直接使用项目根目录
+        if multi_agent_mode and not relative_dir:
+            return self.project_path.resolve()
         if not relative_dir:
             raise ValueError("交付目录不能为空，必须指定")
         deliverables_path = (self.project_path / relative_dir).resolve()

modules/sub_agent/manager.py

@@ -136,7 +136,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         agent_id: int,
         summary: str,
         task: str,
-        deliverables_dir: str,
+        deliverables_dir: Optional[str] = None,
         timeout_seconds: Optional[int] = None,
         conversation_id: Optional[str] = None,
         run_in_background: bool = False,
@@ -145,6 +145,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         multi_agent_mode: bool = False,
         role_id: Optional[str] = None,
         display_name: Optional[str] = None,
+        system_prompt: Optional[str] = None,
+        task_message: Optional[str] = None,
     ) -> Dict:
         """创建子智能体任务并启动协程。
         
@@ -152,7 +154,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         参数 role_id: 多智能体模式下的角色标诶。
         参数 display_name: 多智能体模式下的显示名（如 UI Operator_1）。
         """
-        validation_error = self._validate_create_params(agent_id, summary, task, deliverables_dir)
+        validation_error = self._validate_create_params(agent_id, summary, task, deliverables_dir, multi_agent_mode=multi_agent_mode)
         if validation_error:
             return {"success": False, "error": validation_error}
 
@@ -181,7 +183,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         task_root.mkdir(parents=True, exist_ok=True)
 
         try:
-            deliverables_path = self._resolve_deliverables_dir(deliverables_dir)
+            deliverables_path = self._resolve_deliverables_dir(deliverables_dir, multi_agent_mode=multi_agent_mode)
         except ValueError as exc:
             return {"success": False, "error": str(exc)}
 
@@ -194,11 +196,17 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
 
         prompt_workspace = self._get_runtime_path(self.project_path)
         deliverables_display = self._get_runtime_path(deliverables_path)
+        if task_message:
+            user_message = task_message
+        else:
             user_message = build_user_message(agent_id, summary, task, deliverables_display, timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT)
         task_file.write_text(user_message, encoding="utf-8")
 
-        system_prompt = build_system_prompt(prompt_workspace)
+        if system_prompt:
-        system_prompt_file.write_text(system_prompt, encoding="utf-8")
+            final_system_prompt = system_prompt
+        else:
+            final_system_prompt = build_system_prompt(prompt_workspace)
+        system_prompt_file.write_text(final_system_prompt, encoding="utf-8")
 
         timeout_seconds = timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT
 
@@ -250,7 +258,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
             manager=self,
             task_record=task_record,
             task_message=user_message,
-            system_prompt=system_prompt,
+            system_prompt=final_system_prompt,
             model_key=model_key,
             thinking_mode=thinking_mode,
             multi_agent_mode=multi_agent_mode,
@@ -360,6 +368,11 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         task_id = task["task_id"]
         running_task = self._running_tasks.pop(task_id, None)
         if running_task and not running_task.done():
+            # 子智能体运行在独立事件循环线程中，取消操作必须投递到该循环
+            try:
+                loop = running_task.get_loop()
+                loop.call_soon_threadsafe(running_task.cancel)
+            except Exception:
                 running_task.cancel()
             deadline = time.time() + 5
             while not running_task.done() and time.time() < deadline:
@@ -574,7 +587,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi
         sub_agent = self._find_sub_agent_task_by_agent_id(agent_id)
         if not sub_agent:
             return False
-        sub_agent.messages.append({"role": "user", "content": message_text})
+        sub_agent.inject_message(message_text)
         return True
 
     def _find_sub_agent_task_by_agent_id(self, agent_id: int) -> Optional[Any]:

modules/sub_agent/state.py

@@ -257,6 +257,11 @@ class SubAgentStateMixin:
                     running_task.result(timeout=0)
                 except Exception:
                     pass
+                # 多智能体模式：任务自然进入 idle 时不写输出文件，不应标记为失败
+                if task.get("multi_agent_mode") and not Path(task.get("output_file", "")).exists():
+                    task["status"] = "idle"
+                    task["updated_at"] = time.time()
+                    return {"status": "idle", "task_id": task_id}
                 return self._check_task_status(task)
             return {"status": "running", "task_id": task_id}
 
@@ -264,6 +269,12 @@ class SubAgentStateMixin:
         if output_file.exists():
             return self._check_task_status(task)
 
+        # 多智能体模式：没有输出文件表示 idle，不强制清理
+        if task.get("multi_agent_mode"):
+            task["status"] = "idle"
+            task["updated_at"] = time.time()
+            return {"status": "idle", "task_id": task_id}
+
         if self._should_force_cleanup_stale_task(task):
             return self._mark_task_terminated(
                 task,

modules/sub_agent/task.py

@@ -4,11 +4,15 @@ import asyncio
 import base64
 import json
 import mimetypes
+import re
+import threading
 import time
 import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Set, TYPE_CHECKING
+
+_QUESTION_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE)
 
 from modules.sub_agent.toolkit import (
     SUB_AGENT_TOOLS,
@@ -90,8 +94,14 @@ class SubAgentTask:
         # 多智能体模式相关字段
         self.multi_agent_mode = bool(multi_agent_mode)
         self.multi_agent_state = multi_agent_state
-        # display_name 不传时回退为 'Agent_{agent_id}'
+        # display_name 不传时回退为 'Agent_{self.agent_id}'
         self.display_name = display_name or f"Agent_{self.agent_id}"
+        # 多智能体运行期控制
+        # 使用 threading.Event 避免跨事件循环唤醒问题
+        self._continue_event = threading.Event()
+        self._idle = False
+        self._pending_answer_question_id: Optional[str] = None
+        self._answered_question_ids: Set[str] = set()
 
     def emit(self, type_: str, data: Dict[str, Any]) -> None:
         """输出一行 JSONL 到 progress 文件并缓存。"""
@@ -123,34 +133,57 @@ class SubAgentTask:
         if self.multi_agent_mode:
             tools = list(SUB_AGENT_TOOLS)
             tools.extend(_load_multi_agent_sub_agent_tools())
-            # 多智能体模式下不要求 finish_task，自然输出结束即本轮任务结束
+            # 多智能体模式下不要求 finish_task，自然输出结束即进入 idle，可继续接收消息
         else:
             tools = list(SUB_AGENT_TOOLS)
             tools.append(FINISH_TOOL)
 
         start_time = time.time()
         max_turns = 50
+        turn = 0
 
-        for turn in range(1, max_turns + 1):
+        while not self._cancelled:
-            if self._cancelled:
-                break
             elapsed = time.time() - start_time
             if elapsed > self.timeout_seconds:
                 await self._write_timeout(elapsed)
                 return
 
+            # 多智能体模式下，idle 时等待新消息或外部回答；超时后继续循环检查
+            if self.multi_agent_mode and self._idle:
+                try:
+                    await asyncio.wait_for(
+                        asyncio.get_event_loop().run_in_executor(None, self._continue_event.wait),
+                        timeout=1.0,
+                    )
+                except asyncio.TimeoutError:
+                    pass
+                self._continue_event.clear()
+                if self._cancelled:
+                    break
+                self._idle = False
+                continue
+
+            turn += 1
+            if turn > max_turns:
+                await self._write_failure("任务执行超过最大轮次限制", max_turns_exceeded=True)
+                return
+
             self.stats["api_calls"] += 1
             self.stats["turn_count"] = turn
             self.stats["runtime_seconds"] = int(elapsed)
             self.emit("stats", {**self.stats, "turn_count": turn})
 
+            # 多智能体模式：在模型调用前识别是否有待回答的提问
+            if self.multi_agent_mode:
+                self._pending_answer_question_id = self._peek_pending_question_id()
+
             assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools)
             if usage:
                 self._apply_usage(usage)
 
             # 多智能体模式：把 assistant 文本输出作为进度/完成 output 转发到主对话
             if self.multi_agent_mode and self.multi_agent_state and assistant_message.strip():
-                self._forward_output_to_master(assistant_message)
+                self._forward_output_to_master(assistant_message, is_final=not tool_calls)
 
             final_message: Dict[str, Any] = {"role": "assistant", "content": assistant_message}
             if reasoning:
@@ -160,10 +193,11 @@ class SubAgentTask:
             self.messages.append(final_message)
 
             if not tool_calls:
-                # 多智能体模式：没有 tool_calls 表示本轮结束，进入 idle 状态
+                # 多智能体模式：没有 tool_calls 表示本轮结束，进入 idle 等待
                 if self.multi_agent_mode:
                     self._mark_idle()
-                    return
+                    self._idle = True
+                    continue
                 # 普通模式：prompt 并要求继续 / finish_task
                 self.messages.append({
                     "role": "user",
@@ -197,15 +231,21 @@ class SubAgentTask:
                     "content": content,
                 })
 
-        await self._write_failure("任务执行超过最大轮次限制", max_turns_exceeded=True)
+        # 循环结束（取消或 idle 被外部终止）后的清理
+        if self.multi_agent_mode and self._cancelled:
+            if self.multi_agent_state:
+                self.multi_agent_state.mark_status(self.agent_id, "terminated")
 
-    def _forward_output_to_master(self, output_text: str) -> None:
+    def _forward_output_to_master(self, output_text: str, *, is_final: bool = False) -> None:
         """把子智能体的 assistant 文本输出转发成主对话的 user 消息。"""
         if not self.multi_agent_state:
             return
+        # 如果这是对 pending 提问的回答，不走主对话转发，而是返回到 ask 工具结果
+        if self._provide_answer(output_text):
+            return
         try:
             from modules.multi_agent.state import build_sub_agent_output_text
-            msg = build_sub_agent_output_text(self.display_name, output_text.strip())
+            msg = build_sub_agent_output_text(self.display_name, output_text.strip(), is_final=is_final)
             self.multi_agent_state.push_master_message(msg)
             # 同时记录到实例状态，供 list_active_sub_agents 使用
             inst = self.multi_agent_state.get_instance(self.agent_id)
@@ -215,10 +255,43 @@ class SubAgentTask:
             logger.warning(f"[SubAgentTask] forward output to master failed: {exc}")
 
     def _mark_idle(self) -> None:
-        """多智能体模式下，子智能体自然结束']=本轮任务 结束，进入 idle 状态。"""
+        """多智能体模式下，子智能体自然结束即本轮任务结束，进入 idle 状态。"""
         if self.multi_agent_state:
             self.multi_agent_state.mark_status(self.agent_id, "idle")
 
+    def inject_message(self, message_text: str) -> None:
+        """外部向子智能体上下文插入 user 消息，并唤醒 idle 状态。"""
+        self.messages.append({"role": "user", "content": message_text})
+        self._continue_event.set()
+
+    def _peek_pending_question_id(self) -> Optional[str]:
+        """检查最后一条 user 消息是否是向本智能体提问，返回 question_id。"""
+        if not self.multi_agent_mode or not self.messages:
+            return None
+        for msg in reversed(self.messages):
+            if msg.get("role") == "user":
+                content = msg.get("content") or ""
+                if "的提问" in content:
+                    m = _QUESTION_ID_RE.search(content)
+                    if m:
+                        qid = m.group(1)
+                        if qid not in self._answered_question_ids:
+                            return qid
+                break
+        return None
+
+    def _provide_answer(self, output_text: str) -> bool:
+        """如果当前输出是对 pending 提问的回答，把回答写回 future 并阻止转发到主对话。"""
+        if not self._pending_answer_question_id or not self.multi_agent_state:
+            return False
+        self.multi_agent_state.provide_answer(
+            self._pending_answer_question_id,
+            output_text.strip(),
+        )
+        self._answered_question_ids.add(self._pending_answer_question_id)
+        self._pending_answer_question_id = None
+        return True
+
     def _build_client(self) -> tuple:
         """加载模型配置并初始化 DeepSeekClient。"""
         config_path = self.manager.models_config_file

modules/sub_agent/toolkit.py

@@ -6,6 +6,8 @@ import json
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+from config.model_profiles import _parse_env_ref
+
 # 子智能体可用工具定义（与前端进度展示兼容）
 SUB_AGENT_TOOLS: List[Dict[str, Any]] = [
     {
@@ -267,8 +269,8 @@ def _format_tool_result(name: str, raw: Any) -> str:
 def _build_sub_agent_profile(model_raw: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     """把 sub_agent_models.json 中的模型条目转成 DeepSeekClient.apply_profile 所需格式。"""
     name = str(model_raw.get("name") or model_raw.get("model_name") or model_raw.get("model") or "").strip()
-    url = str(model_raw.get("url") or model_raw.get("base_url") or "").strip()
+    url = str(_parse_env_ref(model_raw.get("url") or model_raw.get("base_url") or "") or "").strip()
-    api_key = str(model_raw.get("apikey") or model_raw.get("api_key") or "").strip()
+    api_key = str(_parse_env_ref(model_raw.get("apikey") or model_raw.get("api_key") or "") or "").strip()
     if not name or not url or not api_key:
         return None
 

server/chat_flow_task_main.py

@@ -130,7 +130,7 @@ from .chat_flow_runtime import (
     detect_malformed_tool_call,
 )
 
-from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates
+from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages
 from .chat_flow_tool_loop import execute_tool_calls
 from .chat_flow_stream_loop import run_streaming_attempts
 from .deep_compression import run_deep_compression
@@ -778,6 +778,25 @@ def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) -
                 "sort_key": update.get("updated_at") or time.time(),
             })
 
+    # 3) 多智能体模式：把子智能体转发到主对话的 pending 消息也作为通知池项消费
+    if getattr(web_terminal, "multi_agent_mode", False):
+        sub_manager = getattr(web_terminal, "sub_agent_manager", None)
+        if sub_manager:
+            state = sub_manager.get_multi_agent_state(conversation_id)
+            if state:
+                ma_messages = state.drain_master_messages()
+                for msg_text in ma_messages:
+                    notices.append({
+                        "kind": "multi_agent",
+                        "message": msg_text,
+                        "payload": {
+                            "sub_agent_notice": True,
+                            "message_source": "sub_agent",
+                            "multi_agent_output": True,
+                        },
+                        "sort_key": time.time(),
+                    })
+
     notices.sort(key=lambda item: item.get("sort_key") or 0)
     return notices
 
@@ -787,18 +806,37 @@ def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool:
     sub_manager = getattr(web_terminal, "sub_agent_manager", None)
     if sub_manager:
         announced = getattr(web_terminal, "_announced_sub_agent_tasks", set())
+        has_running_non_ma = False
+        has_unnotified_non_ma = False
+        has_running_ma = False
         for task in sub_manager.tasks.values():
             if not isinstance(task, dict):
                 continue
-            if not task.get("run_in_background"):
-                continue
             if task.get("conversation_id") != conversation_id:
                 continue
             status = task.get("status")
+            multi_agent_flag = task.get("multi_agent_mode") or False
             if status not in TERMINAL_STATUSES.union({"terminated"}):
-                return True  # 仍在运行
+                if multi_agent_flag:
-            if status != "terminated" and (task.get("task_id") not in announced) and not task.get("notified"):
+                    has_running_ma = True
-                return True  # 已完成但未通知
+                elif task.get("run_in_background"):
+                    has_running_non_ma = True
+                continue
+            if not multi_agent_flag and task.get("run_in_background") and (task.get("task_id") not in announced) and not task.get("notified"):
+                has_unnotified_non_ma = True
+        if has_running_non_ma or has_unnotified_non_ma:
+            return True
+        # 多智能体模式：有未消费的主对话消息 或 有运行中（非 idle）实例时继续轮询
+        if getattr(web_terminal, "multi_agent_mode", False) and has_running_ma:
+            state = sub_manager.get_multi_agent_state(conversation_id)
+            if state:
+                if state.has_pending_master_messages():
+                    return True
+                # 只要还有非 idle 实例就继续；全部 idle 且无 pending 则结束轮询
+                if any(a.status not in {"idle", "terminated"} for a in state.list_all()):
+                    return True
+                return False
+            return True
     bg_manager = getattr(web_terminal, "background_command_manager", None)
     if bg_manager:
         try:
@@ -854,6 +892,11 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation
             await asyncio.sleep(1)
             continue
 
+        # 多智能体模式：主对话任务仍在运行时，由主循环自己消费 pending 消息
+        if getattr(web_terminal, "_multi_agent_main_task_active", False):
+            await asyncio.sleep(1)
+            continue
+
         notices = _collect_pending_completion_notices(
             web_terminal=web_terminal,
             conversation_id=conversation_id,
@@ -928,6 +971,9 @@ async def handle_task_with_sender(
 
     web_terminal = terminal
     conversation_id = getattr(web_terminal.context_manager, "current_conversation_id", None)
+    # 多智能体模式：标记主对话任务正在运行，供后台通知池判断是否可以安全消费
+    if getattr(web_terminal, "multi_agent_mode", False):
+        web_terminal._multi_agent_main_task_active = True
     videos = videos or []
     raw_sender = sender
 
@@ -1638,6 +1684,22 @@ async def handle_task_with_sender(
                         debug_log(f"[Goal] 目标停止：{goal_result.get('reason')}")
             except Exception as exc:
                 debug_log(f"[Goal] turn 结束处理失败: {exc}")
+
+            # 多智能体模式：没有工具调用时，先消费子智能体待转发到主对话的消息；
+            # 如果有新消息注入，继续迭代让 Team Leader 响应，而不是直接结束任务。
+            if getattr(web_terminal, "multi_agent_mode", False):
+                injected_count = await process_multi_agent_master_messages(
+                    messages=messages,
+                    inline=False,
+                    web_terminal=web_terminal,
+                    sender=sender,
+                    debug_log=debug_log,
+                )
+                if injected_count:
+                    debug_log(f"[MultiAgent] no-tool-call turn 注入 {injected_count} 条子智能体消息，继续迭代")
+                    is_first_iteration = False
+                    continue
+
             break
 
         # 目标模式：本轮段确实产生了工具调用
@@ -1699,6 +1761,7 @@ async def handle_task_with_sender(
             last_tool_call_time=last_tool_call_time,
             process_sub_agent_updates=process_sub_agent_updates,
             process_background_command_updates=process_background_command_updates,
+            process_multi_agent_master_messages=process_multi_agent_master_messages,
             maybe_mark_failure_from_message=maybe_mark_failure_from_message,
             mark_force_thinking=mark_force_thinking,
             get_stop_flag=get_stop_flag,
@@ -1749,8 +1812,22 @@ async def handle_task_with_sender(
         # 标记不再是第一次迭代
         is_first_iteration = False
 
+        # 多智能体模式：在进入下一轮模型调用前，消费子智能体最新输出到主对话
+        if getattr(web_terminal, "multi_agent_mode", False):
+            await process_multi_agent_master_messages(
+                messages=messages,
+                inline=False,
+                web_terminal=web_terminal,
+                sender=sender,
+                debug_log=debug_log,
+            )
+
     
     # 最终统计
+    # 多智能体模式：主对话任务结束，解除活跃标记
+    if getattr(web_terminal, "multi_agent_mode", False):
+        web_terminal._multi_agent_main_task_active = False
+
     debug_log(f"\n{'='*40}")
     debug_log(f"任务完成统计:")
     debug_log(f"  总迭代次数: {total_iterations}")

server/chat_flow_task_support.py

@@ -314,6 +314,110 @@ async def process_background_command_updates(*, messages: List[Dict], inline: bo
 
 
 
+def inject_multi_agent_master_message(
+    *,
+    web_terminal,
+    messages,
+    text: str,
+    sender,
+    conversation_id: Optional[str] = None,
+    inline: bool = True,
+    after_tool_call_id: Optional[str] = None,
+) -> Optional[str]:
+    """把多智能体子智能体输出/消息以原生格式注入主对话，不添加 [系统通知|xxx] 前缀。"""
+    raw = "" if text is None else str(text).strip()
+    if not raw:
+        return None
+
+    metadata = {
+        "runtime_injected": True,
+        "source": "sub_agent",
+        "message_source": "sub_agent",
+        "inline": inline,
+        "is_auto_generated": True,
+        "auto_message_type": "multi_agent_output",
+        "visibility": "compact",
+        "starts_work": False,
+    }
+
+    try:
+        ctx_manager = getattr(web_terminal, "context_manager", None)
+        if ctx_manager is not None:
+            ctx_manager.add_conversation("user", raw, metadata=metadata)
+    except Exception:
+        pass
+
+    if messages is not None:
+        insert_index = len(messages)
+        if after_tool_call_id:
+            for idx, msg in enumerate(messages):
+                if msg.get("role") == "tool" and msg.get("tool_call_id") == after_tool_call_id:
+                    end = idx + 1
+                    while end < len(messages) and messages[end].get("role") == "tool":
+                        end += 1
+                    insert_index = end
+                    break
+        messages.insert(insert_index, {"role": "user", "content": raw})
+
+    if callable(sender):
+        payload = {
+            "message": raw,
+            "content": raw,
+            "conversation_id": conversation_id,
+            "inline": inline,
+            "source": "sub_agent",
+            "message_source": "sub_agent",
+            "visibility": "compact",
+            "starts_work": False,
+            "metadata": metadata,
+            "runtime_injected": True,
+        }
+        try:
+            sender("user_message", payload)
+        except Exception:
+            pass
+
+    return raw
+
+
+async def process_multi_agent_master_messages(
+    *,
+    web_terminal,
+    messages,
+    sender,
+    debug_log,
+    inline: bool = False,
+    after_tool_call_id: Optional[str] = None,
+) -> int:
+    """从 MultiAgentState 取出待插入主对话的消息并注入。返回注入条数。"""
+    if not getattr(web_terminal, "multi_agent_mode", False):
+        return 0
+    manager = getattr(web_terminal, "sub_agent_manager", None)
+    if not manager:
+        return 0
+    conversation_id = getattr(getattr(web_terminal, "context_manager", None), "current_conversation_id", None)
+    if not conversation_id:
+        return 0
+    state = manager.get_multi_agent_state(conversation_id)
+    if not state:
+        return 0
+    pending = state.drain_master_messages()
+    if not pending:
+        return 0
+    debug_log(f"[MultiAgent] draining {len(pending)} pending master messages")
+    for msg in pending:
+        inject_multi_agent_master_message(
+            web_terminal=web_terminal,
+            messages=messages,
+            text=msg,
+            sender=sender,
+            conversation_id=conversation_id,
+            inline=inline,
+            after_tool_call_id=after_tool_call_id,
+        )
+    return len(pending)
+
+
 async def wait_retry_delay(*, delay_seconds: int, client_sid: str, username: str, sender, get_stop_flag, clear_stop_flag) -> bool:
     """等待重试间隔，同时检查是否收到停止请求。"""
     if delay_seconds <= 0:

server/chat_flow_tool_loop.py

@@ -24,7 +24,7 @@ from modules.personalization_manager import load_personalization_config, resolve
 from modules.auto_approval_service import run_auto_approval
 from modules.user_question_manager import format_user_question_answer
 from .deep_compression import run_deep_compression
-from .chat_flow_task_support import inject_runtime_user_message
+from .chat_flow_task_support import inject_runtime_user_message, process_multi_agent_master_messages
 
 
 def _format_numbered_lines(lines: List[str], start_line_no: int) -> List[Dict[str, Any]]:
@@ -261,7 +261,7 @@ async def _wait_for_user_questions(*, question_ids: List[str], username: str, ti
     return answered
 
 
-async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None):
+async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages=process_multi_agent_master_messages, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None):
     previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False)
     web_terminal._tool_loop_active = True
     allowed_tool_names = set()
@@ -1323,6 +1323,14 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie
             debug_log=debug_log,
             maybe_mark_failure_from_message=maybe_mark_failure_from_message,
         )
+        await process_multi_agent_master_messages(
+            messages=messages,
+            inline=True,
+            after_tool_call_id=last_completed_tool_call_id,
+            web_terminal=web_terminal,
+            sender=sender,
+            debug_log=debug_log,
+        )
 
     # 运行期模式通知：必须等待同一轮全部 tool_call 都完成后再注入，
     # 避免在 assistant.tool_calls 与对应 tool 消息之间插入 user 消息导致 API 报错。

server/multi_agent.py

@@ -13,19 +13,47 @@ from typing import Any, Dict, List
 
 from flask import Blueprint, current_app, jsonify, request, session
 
-from server.auth_helpers import api_login_required, get_current_username
+from server.auth_helpers import api_login_required, login_required, get_current_username
 from server.context import get_user_resources
 
 multi_agent_bp = Blueprint("multi_agent", __name__)
 
 
 @multi_agent_bp.route("/multiagent/new")
-@api_login_required
+@login_required
 def multi_agent_new_page():
     """多智能体模式入口，返回与 /new 相同的 SPA index.html。"""
     return current_app.send_static_file("index.html")
 
 
+@multi_agent_bp.route("/multiagent/<path:conversation_id>")
+@login_required
+def multi_agent_conversation_page(conversation_id: str):
+    """多智能体模式 指定会话 URL，返回 SPA index.html 让前端路由处理。"""
+    return current_app.send_static_file("index.html")
+
+
+@multi_agent_bp.route("/api/multiagent/rebuild-index", methods=["POST"])
+@api_login_required
+def rebuild_conversation_index_api():
+    """强制从磁盘重建对话索引，补全 multi_agent_mode 等新字段。"""
+    try:
+        username = get_current_username()
+        if not username:
+            return jsonify({"success": False, "error": "未登录"}), 401
+        terminal, _ = get_user_resources(username)
+        if not terminal:
+            return jsonify({"success": False, "error": "工作区未就绪"}), 503
+        cm = getattr(getattr(terminal, "context_manager", None), "conversation_manager", None)
+        if not cm:
+            return jsonify({"success": False, "error": "对话管理器未初始化"}), 503
+        rebuilt = cm._rebuild_index_from_files()
+        cm._save_index(rebuilt)
+        return jsonify({"success": True, "index_size": len(rebuilt)})
+    except Exception as exc:
+        return jsonify({"success": False, "error": str(exc)}), 500
+
+
 @multi_agent_bp.route("/api/multiagent/roles", methods=["GET"])
 @api_login_required
 def list_roles_api():

static/src/app/methods/conversation/load.ts

@@ -180,6 +180,9 @@ export const loadMethods = {
         if (typeof result.model_key === 'string' && result.model_key) {
           this.modelSet(result.model_key);
         }
+        if (typeof result.multi_agent_mode === 'boolean') {
+          this.multiAgentMode = result.multi_agent_mode;
+        }
 
         // 2. 更新当前对话信息
         this.skipConversationHistoryReload = true;
@@ -193,10 +196,11 @@ export const loadMethods = {
         if (!preserveListPosition) {
           this.promoteConversationToTop(conversationId);
         }
+        const urlPrefix = this.multiAgentMode ? '/multiagent/' : '/';
         history.pushState(
           { conversationId },
           '',
-          `/${this.stripConversationPrefix(conversationId)}`
+          `${urlPrefix}${this.stripConversationPrefix(conversationId)}`
         );
         this.skipConversationLoadedEvent = true;
 

static/src/app/methods/ui/route.ts

@@ -41,6 +41,12 @@ export const routeMethods = {
       this.startTitleTyping('多智能体模式', { animate: false });
       this.initialRouteResolved = true;
       this.refreshBlankHeroState();
+      // 进入多智能体模式时触发后端重建索引以补全 multi_agent_mode 字段
+      try {
+        await fetch('/api/multiagent/rebuild-index', { method: 'POST' });
+      } catch (_e) {
+        // 重建失败不阻断主流程
+      }
       // 多智能体模式下自动创建一个带 metadata.multi_agent_mode=true 的新对话
       try {
         const resp = await fetch('/api/multiagent/conversations', {

static/src/app/watchers.ts

@@ -1,7 +1,11 @@
 // @ts-nocheck
 import { debugLog, traceLog } from './methods/common';
+import { useConversationStore } from '../stores/conversation';
 
 export const watchers = {
+  multiAgentMode(newValue) {
+    useConversationStore().$patch({ multiAgentMode: !!newValue });
+  },
   inputMessage() {
     this.autoResizeInput();
     if (typeof this.scheduleComposerDraftPersist === 'function') {

static/src/auth/LoginApp.vue

@@ -69,7 +69,7 @@ const submitting = ref(false);
 const hostSubmitting = ref(false);
 const hostModeEnabled = ref(false);
 
-const login = async () => {
+const doLogin = async (redirectUrl = '/') => {
   if (!email.value || !password.value) {
     error.value = '请输入邮箱和密码';
     return;
@@ -92,7 +92,7 @@ const login = async () => {
 
     const data = await resp.json();
     if (data.success) {
-      window.location.href = '/';
+      window.location.href = redirectUrl;
       return;
     }
 
@@ -104,12 +104,9 @@ const login = async () => {
   }
 };
 
-const enterMultiAgent = () => {
+const login = () => doLogin('/');
-  // 直接跳转，后端 @login_required 会确保已登录后重定向回来
-  window.location.href = '/multiagent/new';
-};
 
-const hostLogin = async () => {
+const doHostLogin = async (redirectUrl = '/') => {
   hostSubmitting.value = true;
   error.value = '';
 
@@ -122,7 +119,7 @@ const hostLogin = async () => {
 
     const data = await resp.json();
     if (data.success) {
-      window.location.href = '/';
+      window.location.href = redirectUrl;
       return;
     }
 
@@ -134,6 +131,20 @@ const hostLogin = async () => {
   }
 };
 
+const hostLogin = () => doHostLogin('/');
+
+const enterMultiAgent = () => {
+  if (hostModeEnabled.value) {
+    doHostLogin('/multiagent/new');
+    return;
+  }
+  if (!email.value || !password.value) {
+    error.value = '请输入邮箱和密码，或选择宿主机模式';
+    return;
+  }
+  doLogin('/multiagent/new');
+};
+
 onMounted(async () => {
   applyTheme(loadTheme());
 

static/src/stores/conversation.ts

@@ -45,6 +45,7 @@ interface ConversationState {
   runningWorkspaceTasks: any[];
   acknowledgedCompletedTaskIds: string[];
   workspaceGroups: WorkspaceConversationGroup[];
+  multiAgentMode: boolean;
 }
 
 export const useConversationStore = defineStore('conversation', {
@@ -70,7 +71,8 @@ export const useConversationStore = defineStore('conversation', {
     conversationsLimit: 20,
     runningWorkspaceTasks: [],
     acknowledgedCompletedTaskIds: [],
-    workspaceGroups: []
+    workspaceGroups: [],
+    multiAgentMode: false
   }),
   actions: {
     resetConversations() {
@@ -192,8 +194,9 @@ export const useConversationStore = defineStore('conversation', {
       const fetchOffset = refresh ? 0 : group.offset;
       group.loading = true;
       try {
+        const maParam = this.multiAgentMode ? '&multi_agent_mode=1' : '&multi_agent_mode=0';
         const response = await fetch(
-          `/api/conversations?workspace_id=${encodeURIComponent(workspaceId)}&limit=${group.fetchLimit}&offset=${fetchOffset}`
+          `/api/conversations?workspace_id=${encodeURIComponent(workspaceId)}&limit=${group.fetchLimit}&offset=${fetchOffset}${maParam}`
         );
         const data = await response.json();
         if (data.success) {

utils/conversation_manager/crud_mixin.py

@@ -233,7 +233,8 @@ class CrudMixin:
                 "has_videos": conversation_data["metadata"].get("has_videos", False),
                 "total_messages": metadata.total_messages,
                 "total_tools": metadata.total_tools,
-                "status": metadata.status
+                "status": metadata.status,
+                "multi_agent_mode": bool(conversation_data["metadata"].get("multi_agent_mode", False))
             }
             
             self._save_index(index)

utils/conversation_manager/index_mixin.py

@@ -163,6 +163,7 @@ class IndexMixin:
                 "total_messages": metadata.get("total_messages", 0),
                 "total_tools": metadata.get("total_tools", 0),
                 "status": metadata.get("status", "active"),
+                "multi_agent_mode": bool(metadata.get("multi_agent_mode", False)),
             }
         elapsed_ms = (time.perf_counter() - t0) * 1000
         if perf_log: