From 7f2ad9144d2ef71ad4aaf5a8652a7d08e7c6391b Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Sun, 12 Jul 2026 18:39:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(multi-agent):=20=E4=BF=AE=E5=A4=8D=E5=AD=90?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E7=8A=B6=E6=80=81=E8=AF=AF=E5=88=A4?= =?UTF-8?q?=E3=80=81=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E4=B8=8E=20idle=20=E7=AD=89=E5=BE=85=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/__init__.py | 7 +- core/main_terminal_parts/tools_execution.py | 57 +++++- core/web_terminal.py | 29 ++- modules/multi_agent/debug_logger.py | 30 +++ modules/multi_agent/state.py | 29 ++- modules/multi_agent/tools.py | 5 +- modules/sub_agent/manager.py | 76 +++++++- modules/sub_agent/state.py | 72 ++++++- modules/sub_agent/task.py | 179 ++++++++++++++++-- server/chat_flow_task_main.py | 30 ++- server/chat_flow_task_support.py | 15 ++ server/conversation.py | 8 +- static/src/app/methods/conversation/action.ts | 12 +- static/src/app/methods/message/send.ts | 15 +- .../overlay/SubAgentActivityDialog.vue | 175 +++++++++++++---- static/src/composables/useLegacySocket.ts | 22 ++- 16 files changed, 659 insertions(+), 102 deletions(-) create mode 100644 modules/multi_agent/debug_logger.py diff --git a/config/__init__.py b/config/__init__.py index bfdf7ef..98f68e7 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -16,11 +16,13 @@ def _load_dotenv(): import sys pre_existing_keys = set(os.environ.keys()) - # 1) 仓库根 .env（开发便利，不覆盖已有环境变量） + # 1) 仓库根 .env（开发便利，不覆盖已有的环境变量，但 ASTRION_DATA_ROOT + # 作为项目数据根目录必须优先以 .env 为准，避免外部 shell 误指到 clone） if getattr(sys, 'frozen', False): env_path = Path.home() / '.astrion' / 'astrion' / '.env' else: env_path = Path(__file__).resolve().parents[1] / '.env' + env_from_file: dict = {} if env_path.exists(): try: for raw_line in env_path.read_text(encoding="utf-8").splitlines(): @@ -34,11 +36,14 @@ def _load_dotenv(): value = value.strip().strip('"').strip("'") if not key: continue + env_from_file[key] = value if key in pre_existing_keys: continue os.environ[key] = value except Exception: pass + if "ASTRION_DATA_ROOT" in env_from_file: + os.environ["ASTRION_DATA_ROOT"] = str(Path(env_from_file["ASTRION_DATA_ROOT"]).expanduser()) # 2) settings.json（数据根下的统一配置） data_root = os.environ.get("ASTRION_DATA_ROOT", str(Path.home() / ".astrion" / "astrion")) diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 9dbe060..d9d70aa 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -101,6 +101,7 @@ from utils.api_client import DeepSeekClient from utils.context_manager import ContextManager from utils.tool_result_formatter import format_tool_result_for_context from utils.logger import setup_logger +from modules.multi_agent.debug_logger import ma_debug from config.model_profiles import ( get_model_profile, get_model_prompt_replacements, @@ -1128,6 +1129,27 @@ class MainTerminalToolsExecutionMixin: task_ids.append(task.get("task_id")) if missing: result = {"success": False, "error": f"未找到对应子智能体: {missing}"} + elif getattr(self, "multi_agent_mode", False): + # 多智能体模式：子智能体通过自然输出结束，不调用 finish_task。 + # sleep 不再阻塞等待，而是直接返回各实例最近一次输出内容。 + conv_id = self.context_manager.current_conversation_id + ma_state = manager.get_multi_agent_state(conv_id) + outputs = [] + for aid in normalized_ids: + inst = ma_state.get_instance(aid) if ma_state else None + outputs.append({ + "agent_id": aid, + "display_name": inst.display_name if inst else f"Agent_{aid}", + "status": inst.status if inst else "unknown", + "last_output": inst.last_output if inst else "", + }) + result = { + "success": True, + "mode": "wait_sub_agent_ids", + "agent_ids": normalized_ids, + "outputs": outputs, + "message": f"已获取 {len(normalized_ids)} 个子智能体最近一次输出" + } else: wait_results = [] waited_task_ids = [] @@ -1763,7 +1785,7 @@ class MainTerminalToolsExecutionMixin: agent_id=int(agent_id), summary=summary_text, task=arguments.get("task", ""), - run_in_background=bool(arguments.get("run_in_background", True)), + run_in_background=False, timeout_seconds=arguments.get("timeout_seconds"), conversation_id=conv_id, model_key=role.model_key, @@ -1774,8 +1796,8 @@ class MainTerminalToolsExecutionMixin: system_prompt=system_prompt, task_message=task_message, ) - # 在多智能体模式下，主进程 create_sub_agent 总是后台启动， - # 主智能体不需要阻塞等待，而是通过子智能体输出转发拿进度。 + # 在多智能体模式下，子智能体是团队协作成员，不是传统后台任务。 + # run_in_background=False 避免触发后台完成通知轮询，保持主对话输入区可用。 except Exception as exc: logger.exception("[multi_agent] create_sub_agent failed") result = {"success": False, "error": str(exc)} @@ -1865,12 +1887,27 @@ class MainTerminalToolsExecutionMixin: else: # 构造消息文本并插入子对话 text = build_master_message_to_sub_agent(message) + ma_debug( + "tool_send_message_to_sub_agent", + agent_id=agent_id, + raw_message=str(message)[:500], + wrapped_message_preview=text[:500], + conversation_id=conv_id, + ) ok = self.sub_agent_manager.inject_message_to_sub_agent(agent_id, text) if not ok: result = {"success": False, "error": f"子智能体 {agent_id} 不存在或已结束"} else: result = {"success": True, "agent_id": agent_id} + ma_debug( + "tool_send_message_to_sub_agent_result", + agent_id=agent_id, + conversation_id=conv_id, + ok=ok, + result=result, + ) except Exception as exc: + logger.exception("[multi_agent] send_message_to_sub_agent failed") result = {"success": False, "error": str(exc)} elif tool_name == "ask_sub_agent": @@ -1896,6 +1933,14 @@ class MainTerminalToolsExecutionMixin: msg_id=question_id, target=state.get_instance(agent_id).display_name if state.get_instance(agent_id) else f"Agent_{agent_id}", ) + ma_debug( + "tool_ask_sub_agent", + agent_id=agent_id, + question=str(question)[:500], + question_id=question_id, + wrapped_message_preview=text[:500], + conversation_id=conv_id, + ) ok = self.sub_agent_manager.inject_message_to_sub_agent(agent_id, text) if not ok: result = {"success": False, "error": f"子智能体 {agent_id} 不存在"} @@ -1921,6 +1966,12 @@ class MainTerminalToolsExecutionMixin: answer = arguments.get("answer", "") conv_id = self.context_manager.current_conversation_id state = self.sub_agent_manager.get_multi_agent_state(conv_id) + ma_debug( + "tool_answer_sub_agent_question", + question_id=question_id, + answer_preview=str(answer)[:500], + conversation_id=conv_id, + ) if not state: result = {"success": False, "error": "多智能体状态未就绪"} else: diff --git a/core/web_terminal.py b/core/web_terminal.py index 964574d..24ab6ad 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -3,7 +3,7 @@ import json import time from datetime import datetime -from typing import Dict, List, Optional, Callable, TYPE_CHECKING +from typing import Any, Dict, List, Optional, Callable, TYPE_CHECKING import os from core.main_terminal import MainTerminal from utils.logger import setup_logger @@ -105,13 +105,19 @@ class WebTerminal(MainTerminal): # 新增：对话管理相关方法（Web版本） # =========================================== - def create_new_conversation(self, thinking_mode: bool = None, run_mode: Optional[str] = None) -> Dict: + def create_new_conversation( + self, + thinking_mode: bool = None, + run_mode: Optional[str] = None, + metadata_overrides: Optional[Dict[str, Any]] = None, + ) -> Dict: """ 创建新对话（Web版本） Args: thinking_mode: 思考模式，None则使用当前设置 run_mode: 显式的运行模式（fast/thinking/deep） + metadata_overrides: 额外写入对话 metadata 的字段 Returns: Dict: 包含新对话信息 @@ -171,17 +177,21 @@ class WebTerminal(MainTerminal): try: # 先创建新对话。start_new_conversation 会先 save_current_conversation()， # 此时 terminal.model_key 仍是旧对话的模型，避免把旧对话覆盖成默认模型。 + metadata_overrides_merged = { + "permission_mode": self.get_permission_mode(), + "execution_mode": self.get_execution_mode() if hasattr(self, "get_execution_mode") else "sandbox", + "pending_permission_mode": None, + "pending_execution_mode": None, + # frozen_*_prompt 不在创建时预设，由第一次 build_messages 根据当时的实际模式懒加载并冻结 + } + if isinstance(metadata_overrides, dict): + metadata_overrides_merged.update(metadata_overrides) + conversation_id = self.context_manager.start_new_conversation( project_path=self.project_path, thinking_mode=thinking_mode, run_mode=self.run_mode, - metadata_overrides={ - "permission_mode": self.get_permission_mode(), - "execution_mode": self.get_execution_mode() if hasattr(self, "get_execution_mode") else "sandbox", - "pending_permission_mode": None, - "pending_execution_mode": None, - # frozen_*_prompt 不在创建时预设，由第一次 build_messages 根据当时的实际模式懒加载并冻结 - }, + metadata_overrides=metadata_overrides_merged, ) # 新对话创建完成后再应用默认模型（此时旧对话已安全保存）。 @@ -377,6 +387,7 @@ class WebTerminal(MainTerminal): "run_mode": self.run_mode, "thinking_mode": self.thinking_mode, "model_key": getattr(self, "model_key", None), + "multi_agent_mode": bool(getattr(self, "multi_agent_mode", False)), "message": f"对话已加载: {conversation_id}" } else: diff --git a/modules/multi_agent/debug_logger.py b/modules/multi_agent/debug_logger.py new file mode 100644 index 0000000..7ede7e5 --- /dev/null +++ b/modules/multi_agent/debug_logger.py @@ -0,0 +1,30 @@ +"""多智能体模式专用调试日志。 + +把所有子智能体与主智能体之间的消息往来、注入、转发、工具调用记录下来， +便于排查循环/重复输出等问题。日志写入 {LOGS_DIR}/multi_agent_loop.log。 +""" +from __future__ import annotations + +import json +import time +from pathlib import Path +from typing import Any + +from config.paths import LOGS_DIR + + +_LOG_PATH = Path(LOGS_DIR) / "multi_agent_loop.log" + + +def ma_debug(event: str, **kwargs: Any) -> None: + """追加一条结构化调试日志。""" + try: + _LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + payload = {"t": time.time(), "event": event} + for k, v in kwargs.items(): + payload[k] = v + line = json.dumps(payload, ensure_ascii=False, default=str) + with open(_LOG_PATH, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception: + pass diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index 04f9586..4c51f4a 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -25,6 +25,7 @@ from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, TYPE_CHECKING +from modules.multi_agent.debug_logger import ma_debug if TYPE_CHECKING: from modules.sub_agent.task import SubAgentTask @@ -298,7 +299,17 @@ class MultiAgentState: 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}") + old_fut = self.pending_questions[question_id] + try: + old_loop = old_fut.get_loop() + if old_loop.is_closed(): + self.pending_questions.pop(question_id, None) + self.pending_question_loops.pop(question_id, None) + else: + raise RuntimeError(f"question_id 已存在: {question_id}") + except Exception: + self.pending_questions.pop(question_id, None) + self.pending_question_loops.pop(question_id, None) loop = asyncio.get_running_loop() fut: asyncio.Future = loop.create_future() self.pending_questions[question_id] = fut @@ -329,6 +340,12 @@ class MultiAgentState: 返回 True 表示找到等待中的 future；False 表示无等待方或已超时。 支持跨事件循环调用（例如主对话循环回答子智能体循环里的提问）。 """ + ma_debug( + "state_provide_answer", + question_id=question_id, + has_pending=question_id in self.pending_questions, + answer_preview=str(answer)[:300], + ) # 如果 wait_for_answer 还没注册，先把答案暂存 if question_id not in self.pending_questions: self.pending_answers[question_id] = answer @@ -349,8 +366,14 @@ class MultiAgentState: return True except Exception: pass - # 同循环回退 - return asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), asyncio.get_event_loop()).result(timeout=5) + # 同循环回退（future 所属循环可能已关闭，失败时把答案暂存，避免阻塞方永远等不到） + try: + return asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), asyncio.get_event_loop()).result(timeout=5) + except Exception: + self.pending_questions.pop(question_id, None) + self.pending_question_loops.pop(question_id, None) + self.pending_answers[question_id] = answer + return True def is_agent_blocking(self, agent_id: int) -> bool: return agent_id in self.agent_blocking_question diff --git a/modules/multi_agent/tools.py b/modules/multi_agent/tools.py index bd055e2..b002e61 100644 --- a/modules/multi_agent/tools.py +++ b/modules/multi_agent/tools.py @@ -53,7 +53,10 @@ def _master_tool_create_sub_agent() -> Dict[str, Any]: "type": "integer", "description": "（可选）手动指定实例编号；不传时自动递增。", }, - "timeout_seconds": {"type": "integer", "description": "超时秒数，默认 600。"}, + "timeout_seconds": { + "type": "integer", + "description": "（可选）超时秒数。不填表示该子智能体不会被时间终结，适用于可能多轮长期任务的子智能体；一般情况下推荐创建永久子智能体。", + }, "thinking_mode": { "type": "string", "enum": ["fast", "thinking"], diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index b3139c3..1bcaa3f 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -30,6 +30,7 @@ from modules.sub_agent.tools import handle_search_workspace, handle_read_mediafi from modules.sub_agent.state import SubAgentStateMixin from modules.sub_agent.stats import SubAgentStatsMixin from modules.sub_agent.creation import SubAgentCreationMixin +from modules.multi_agent.debug_logger import ma_debug if TYPE_CHECKING: from core.web_terminal import WebTerminal @@ -199,7 +200,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi 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) + display_timeout = timeout_seconds if timeout_seconds is not None else 0 + user_message = build_user_message(agent_id, summary, task, deliverables_display, display_timeout or SUB_AGENT_DEFAULT_TIMEOUT) task_file.write_text(user_message, encoding="utf-8") if system_prompt: @@ -208,8 +210,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi 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 - + # timeout_seconds 为 None 表示永久子智能体（不会被时间终结） task_record = { "task_id": task_id, "agent_id": agent_id, @@ -223,6 +224,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi "updated_at": time.time(), "conversation_id": conversation_id, "run_in_background": run_in_background, + "multi_agent_mode": bool(multi_agent_mode), "task_root": str(task_root), "output_file": str(output_file), "stats_file": str(stats_file), @@ -286,6 +288,15 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi if multi_agent_mode and display_name: message = f"{display_name} 已创建，任务ID: {task_id}" print(f"{OUTPUT_FORMATS['info']} {message}") + ma_debug( + "manager_create_sub_agent", + task_id=task_id, + agent_id=agent_id, + display_name=display_name, + multi_agent_mode=multi_agent_mode, + run_in_background=task_record.get("run_in_background"), + timeout_seconds=timeout_seconds, + ) return { "success": True, @@ -577,6 +588,26 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi """删除会话状态（会话结束时调用）。""" self.multi_agent_states.pop(conversation_id, None) + def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int: + """修正运行态任务状态。 + + 在父类实现前先根据内存中的 MultiAgentState 给旧任务补上 multi_agent_mode + 标记，避免任务记录缺字段导致被当成普通子智能体误判为 failed。 + """ + if conversation_id and conversation_id in self.multi_agent_states: + state = self.multi_agent_states[conversation_id] + agent_ids = {a.agent_id for a in state.list_all()} + for task in self.tasks.values(): + if ( + isinstance(task, dict) + and task.get("conversation_id") == conversation_id + and task.get("agent_id") in agent_ids + and task.get("multi_agent_mode") is None + ): + task["multi_agent_mode"] = True + task["updated_at"] = time.time() + return super().reconcile_task_states(conversation_id=conversation_id) + def inject_message_to_sub_agent(self, agent_id: int, message_text: str) -> bool: """同事件循环中向子智能体上下文插入 user 消息。 @@ -585,6 +616,13 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi """ # 查找该 agent_id 对应的 running SubAgentTask sub_agent = self._find_sub_agent_task_by_agent_id(agent_id) + ma_debug( + "manager_inject_message_to_sub_agent", + agent_id=agent_id, + message_preview=str(message_text)[:500], + found=bool(sub_agent), + task_id=sub_agent.task_id if sub_agent else None, + ) if not sub_agent: return False sub_agent.inject_message(message_text) @@ -606,18 +644,40 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi def _on_multi_agent_task_done(self, task_id: str, agent_id: int, state: Any, sub_agent: Any) -> None: """SubAgentTask 结束回调会调这个更新 MultiAgentState 实例状态。""" - # 取出当前 status（由 _finalize_task 设置） final_task = self.tasks.get(task_id) or {} - final_status = final_task.get("status") or "terminated" - # 允许的自然退出：running -> idle（未走 _finalize） vs failed/timeout - # SubAgentTask 在多智能体模式下没有 _finalize_task，我们手动赋值 idle - # 如果状态被结 _finalize 为 failed/timeout，则保持该状态 + final_status_before = final_task.get("status") + ma_debug( + "manager_on_multi_agent_task_done", + task_id=task_id, + agent_id=agent_id, + sub_agent_idle=getattr(sub_agent, "_idle", False), + sub_agent_cancelled=getattr(sub_agent, "_cancelled", False), + task_status_before=final_status_before, + ) + # 多智能体模式下，子智能体自然进入 idle 后 Task 可能被外部事件循环取消， + # 或者 reconcile 把 idle 误判为 failed。优先以 SubAgentTask 自身状态为准： + # - 被手动取消 -> terminated + # - 自然进入 idle -> idle（可继续接收消息） + # - 真正异常/超时/finish_task 失败 -> failed/timeout + if getattr(sub_agent, "_cancelled", False): + state.mark_status(agent_id, "terminated") + ma_debug("manager_ma_state_set", agent_id=agent_id, status="terminated", reason="sub_agent_cancelled") + return + if getattr(sub_agent, "_idle", False): + state.mark_status(agent_id, "idle") + ma_debug("manager_ma_state_set", agent_id=agent_id, status="idle", reason="sub_agent_idle") + return + + # 兜底：取出当前 task status（由 _finalize_task 设置） if final_status in TERMINAL_STATUSES: state.mark_status(agent_id, final_status, last_output=str(final_task.get("final_result") or "")) + ma_debug("manager_ma_state_set", agent_id=agent_id, status=final_status, reason="task_terminal_status") elif final_status == "terminated": state.mark_status(agent_id, "terminated") + ma_debug("manager_ma_state_set", agent_id=agent_id, status="terminated", reason="task_terminated_status") else: state.mark_status(agent_id, "idle") + ma_debug("manager_ma_state_set", agent_id=agent_id, status="idle", reason="fallback_idle") def _get_runtime_path(self, host_path: Path) -> str: """将宿主机路径映射为容器内路径（仅用于提示展示）。""" diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index 181668c..0b4467a 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional from utils.logger import setup_logger +from modules.multi_agent.debug_logger import ma_debug logger = setup_logger(__name__) TERMINAL_STATUSES = {"completed", "failed", "timeout"} @@ -94,6 +95,7 @@ class SubAgentStateMixin: try: output = json.loads(output_file.read_text(encoding="utf-8")) except Exception as exc: + logger.warning(f"[check_task_status] output 文件解析失败: {exc}") task["status"] = "failed" task["updated_at"] = time.time() return {"success": False, "status": "failed", "task_id": task_id, "message": f"输出文件解析失败: {exc}"} @@ -103,6 +105,24 @@ class SubAgentStateMixin: stats = output.get("stats", {}) elapsed_seconds = self._compute_elapsed_seconds(task) + ma_debug( + "check_task_status", + task_id=task_id, + agent_id=task.get("agent_id"), + multi_agent_mode=task.get("multi_agent_mode"), + output_status=output.get("status"), + output_success=success, + ) + + # 多智能体模式：output 中 status 为 running/idle 表示子智能体仍在运行或 + # 本轮结束但上下文保留、可继续接收消息，都不是失败/完成。 + # 此时不生成 final_result，只保持原状态，避免 reconcile 把运行中误判为失败。 + if task.get("multi_agent_mode") and output.get("status") in {"running", "idle"}: + task["status"] = output["status"] + task["updated_at"] = time.time() + ma_debug("check_task_status_keep_alive", task_id=task_id, status=output["status"]) + return {"status": output["status"], "task_id": task_id} + if output.get("timeout"): status = "timeout" elif output.get("max_turns_exceeded"): @@ -113,6 +133,7 @@ class SubAgentStateMixin: else: status = "failed" + ma_debug("check_task_status_result", task_id=task_id, agent_id=task.get("agent_id"), status=status) task["status"] = status task["updated_at"] = time.time() if status == "completed" and elapsed_seconds is not None: @@ -237,6 +258,13 @@ class SubAgentStateMixin: if not task: return status = "completed" if success else "failed" + ma_debug( + "mark_task_done", + task_id=task_id, + agent_id=task.get("agent_id"), + success=success, + new_status=status, + ) task["status"] = status task["updated_at"] = time.time() task["runtime_seconds"] = runtime_seconds @@ -246,10 +274,19 @@ class SubAgentStateMixin: def _refresh_task_runtime_state(self, task: Dict[str, Any]) -> Dict[str, Any]: """刷新单个任务运行态。""" status = task.get("status") - if status in TERMINAL_STATUSES.union({"terminated"}): - return {"status": status, "task_id": task.get("task_id")} - task_id = task.get("task_id") + agent_id = task.get("agent_id") + multi_agent_flag = task.get("multi_agent_mode") + ma_debug( + "refresh_task_runtime_state_start", + task_id=task_id, + agent_id=agent_id, + before_status=status, + multi_agent_mode=multi_agent_flag, + ) + if status in TERMINAL_STATUSES.union({"terminated"}): + return {"status": status, "task_id": task_id} + running_task = self._running_tasks.get(task_id) if task_id else None if running_task: if running_task.done(): @@ -261,18 +298,25 @@ class SubAgentStateMixin: if task.get("multi_agent_mode") and not Path(task.get("output_file", "")).exists(): task["status"] = "idle" task["updated_at"] = time.time() + ma_debug("refresh_task_runtime_state_idle_no_output", task_id=task_id) return {"status": "idle", "task_id": task_id} - return self._check_task_status(task) + result = self._check_task_status(task) + ma_debug("refresh_task_runtime_state_done_result", task_id=task_id, result=result) + return result + ma_debug("refresh_task_runtime_state_still_running", task_id=task_id) return {"status": "running", "task_id": task_id} output_file = Path(task.get("output_file", "")) if output_file.exists(): - return self._check_task_status(task) + result = self._check_task_status(task) + ma_debug("refresh_task_runtime_state_output_result", task_id=task_id, result=result) + return result # 多智能体模式：没有输出文件表示 idle，不强制清理 if task.get("multi_agent_mode"): task["status"] = "idle" task["updated_at"] = time.time() + ma_debug("refresh_task_runtime_state_idle_no_output_file", task_id=task_id) return {"status": "idle", "task_id": task_id} if self._should_force_cleanup_stale_task(task): @@ -293,6 +337,7 @@ class SubAgentStateMixin: def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int: """修正运行态任务状态，返回修正条目数。""" changed = 0 + ma_debug("reconcile_task_states_start", conversation_id=conversation_id, task_count=len(self.tasks)) for task in self.tasks.values(): if not isinstance(task, dict): continue @@ -301,10 +346,19 @@ class SubAgentStateMixin: before_status = task.get("status") before_notified = task.get("notified") self._refresh_task_runtime_state(task) - if task.get("status") != before_status or task.get("notified") != before_notified: + after_status = task.get("status") + if after_status != before_status or task.get("notified") != before_notified: + ma_debug( + "reconcile_task_states_changed", + task_id=task.get("task_id"), + agent_id=task.get("agent_id"), + before_status=before_status, + after_status=after_status, + ) changed += 1 if changed: self._save_state() + ma_debug("reconcile_task_states_end", conversation_id=conversation_id, changed=changed) return changed def _should_force_cleanup_stale_task(self, task: Dict[str, Any]) -> bool: @@ -314,8 +368,12 @@ class SubAgentStateMixin: created_at = 0 if created_at <= 0: return False + # timeout_seconds 为 None 表示永久子智能体，不强制清理 + raw_timeout = task.get("timeout_seconds") + if raw_timeout is None: + return False from config import SUB_AGENT_DEFAULT_TIMEOUT - timeout_seconds = int(task.get("timeout_seconds") or SUB_AGENT_DEFAULT_TIMEOUT or 0) + timeout_seconds = int(raw_timeout or SUB_AGENT_DEFAULT_TIMEOUT or 0) timeout_seconds = max(timeout_seconds, 1) grace_seconds = 120 elapsed = time.time() - created_at diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index e90b7e7..779a71e 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -27,6 +27,8 @@ if TYPE_CHECKING: from modules.sub_agent.manager import SubAgentManager from modules.multi_agent.state import MultiAgentState +from modules.multi_agent.debug_logger import ma_debug + logger = setup_logger(__name__) # 多智能体模式下额外加载的工具定义 @@ -63,7 +65,8 @@ class SubAgentTask: self.thinking_mode = thinking_mode or "fast" self.task_id = task_record["task_id"] self.agent_id = task_record["agent_id"] - self.timeout_seconds = int(task_record.get("timeout_seconds") or 180) + raw_timeout = task_record.get("timeout_seconds") + self.timeout_seconds = int(raw_timeout) if raw_timeout is not None else None self.deliverables_dir = Path(task_record["deliverables_dir"]) self.output_file = Path(task_record["output_file"]) @@ -97,8 +100,9 @@ class SubAgentTask: # display_name 不传时回退为 'Agent_{self.agent_id}' self.display_name = display_name or f"Agent_{self.agent_id}" # 多智能体运行期控制 - # 使用 threading.Event 避免跨事件循环唤醒问题 - self._continue_event = threading.Event() + # 使用 asyncio.Event 在子智能体自己的事件循环内等待； + # inject_message 可能跨线程调用，通过 loop.call_soon_threadsafe 唤醒。 + self._continue_event: Optional[asyncio.Event] = None self._idle = False self._pending_answer_question_id: Optional[str] = None self._answered_question_ids: Set[str] = set() @@ -116,16 +120,20 @@ class SubAgentTask: async def run(self) -> None: """主 LLM 循环。""" + # 在子智能体自己的事件循环内初始化 asyncio.Event + self._continue_event = asyncio.Event() try: await self._run_loop() except asyncio.CancelledError: self._cancelled = True logger.debug(f"[SubAgent] task={self.task_id} 被取消") + ma_debug("sub_agent_run_cancelled", task_id=self.task_id, agent_id=self.agent_id, display_name=self.display_name) # shield 避免取消信号中断最终状态落盘 await asyncio.shield(self._write_failure("子智能体被手动终止")) raise except Exception as exc: logger.exception(f"[SubAgent] task={self.task_id} 执行异常") + ma_debug("sub_agent_run_exception", task_id=self.task_id, agent_id=self.agent_id, display_name=self.display_name, error=str(exc)) await self._write_failure(f"执行异常: {exc}") async def _run_loop(self) -> None: @@ -144,22 +152,33 @@ class SubAgentTask: while not self._cancelled: elapsed = time.time() - start_time - if elapsed > self.timeout_seconds: + if self.timeout_seconds is not None and elapsed > self.timeout_seconds: await self._write_timeout(elapsed) return - # 多智能体模式下，idle 时等待新消息或外部回答；超时后继续循环检查 + # 多智能体模式下，idle 时等待新消息或外部回答；只有真正被注入消息时才继续运行 if self.multi_agent_mode and self._idle: + event_set = False try: - await asyncio.wait_for( - asyncio.get_event_loop().run_in_executor(None, self._continue_event.wait), - timeout=1.0, - ) + if self._continue_event is None: + self._continue_event = asyncio.Event() + await asyncio.wait_for(self._continue_event.wait(), timeout=1.0) + event_set = True except asyncio.TimeoutError: pass - self._continue_event.clear() if self._cancelled: break + if not event_set: + # 只是周期性检查取消状态，没有新消息，保持 idle 继续等待 + continue + self._continue_event.clear() + ma_debug( + "sub_agent_idle_wake", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + pending_messages=[m.get("role") for m in self.messages[-3:]], + ) self._idle = False continue @@ -177,6 +196,18 @@ class SubAgentTask: if self.multi_agent_mode: self._pending_answer_question_id = self._peek_pending_question_id() + # 调试：记录进入本轮模型调用前的上下文摘要 + ma_debug( + "sub_agent_model_call_start", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + turn=turn, + message_count=len(self.messages), + last_user_message=self.messages[-1].get("content", "")[:300] if self.messages and self.messages[-1].get("role") == "user" else "", + pending_answer_question_id=self._pending_answer_question_id, + ) + assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools) if usage: self._apply_usage(usage) @@ -191,12 +222,14 @@ class SubAgentTask: if tool_calls: final_message["tool_calls"] = tool_calls self.messages.append(final_message) + self._persist_conversation(partial_summary=assistant_message[:200]) if not tool_calls: # 多智能体模式：没有 tool_calls 表示本轮结束，进入 idle 等待 if self.multi_agent_mode: self._mark_idle() self._idle = True + self._persist_conversation(partial_summary=assistant_message[:200]) continue # 普通模式：prompt 并要求继续 / finish_task self.messages.append({ @@ -230,6 +263,7 @@ class SubAgentTask: "tool_call_id": tool_call.get("id", progress_id), "content": content, }) + self._persist_conversation(partial_summary=assistant_message[:200]) # 循环结束（取消或 idle 被外部终止）后的清理 if self.multi_agent_mode and self._cancelled: @@ -237,7 +271,7 @@ class SubAgentTask: self.multi_agent_state.mark_status(self.agent_id, "terminated") def _forward_output_to_master(self, output_text: str, *, is_final: bool = False) -> None: - """把子智能体的 assistant 文本输出转发成主对话的 user 消息。""" + """把子智能体的 assistant 文本输出转发成主对话的 user 消息，并写入进度文件供前端查看。""" if not self.multi_agent_state: return # 如果这是对 pending 提问的回答，不走主对话转发，而是返回到 ask 工具结果 @@ -251,17 +285,58 @@ class SubAgentTask: inst = self.multi_agent_state.get_instance(self.agent_id) if inst: inst.last_output = output_text[:500] + # 写入进度文件，前端子智能体进度弹窗可直接展示 + self.emit("progress", { + "subtype": "output", + "content": output_text, + "is_final": is_final, + "ts": int(time.time() * 1000), + }) + ma_debug( + "sub_agent_output_forwarded", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + is_final=is_final, + content_preview=output_text[:300], + ) except Exception as exc: logger.warning(f"[SubAgentTask] forward output to master failed: {exc}") def _mark_idle(self) -> None: """多智能体模式下，子智能体自然结束即本轮任务结束，进入 idle 状态。""" + ma_debug( + "sub_agent_mark_idle", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + ) 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}) + ma_debug( + "sub_agent_message_injected", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + message_preview=str(message_text)[:500], + was_idle=self._idle, + ) + # inject_message 可能从其他线程（主对话线程）调用，需要线程安全唤醒。 + # 优先使用子智能体 Task 所属事件循环投递 set()，避免跨线程直接操作 Future。 + if self._continue_event is None: + self._continue_event = asyncio.Event() + if self._task is not None: + try: + task_loop = self._task.get_loop() + if task_loop.is_running(): + task_loop.call_soon_threadsafe(self._continue_event.set) + return + except Exception: + pass self._continue_event.set() def _peek_pending_question_id(self) -> Optional[str]: @@ -409,6 +484,14 @@ class SubAgentTask: if name == "ask_master": question = str(args.get("question") or "").strip() question_id = str(args.get("question_id") or f"ask_master_{uuid.uuid4().hex[:10]}") + ma_debug( + "sub_agent_tool_ask_master", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + question_id=question_id, + question=question[:500], + ) if not question: return {"success": False, "error": "question 不能为空"} # 插入到主对话 @@ -428,6 +511,15 @@ class SubAgentTask: target_id = int(args.get("target_agent_id") or 0) question = str(args.get("question") or "").strip() question_id = str(args.get("question_id") or f"ask_other_{uuid.uuid4().hex[:10]}") + ma_debug( + "sub_agent_tool_ask_other_agent", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + target_agent_id=target_id, + question_id=question_id, + question=question[:500], + ) if not target_id or not question: return {"success": False, "error": "参数缺失"} # 查找目标实例 @@ -516,10 +608,75 @@ class SubAgentTask: async def _write_failure(self, message: str, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None: elapsed = time.time() - (self.stats["runtime_start"] / 1000) + ma_debug( + "sub_agent_write_failure", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + message=message, + max_turns_exceeded=max_turns_exceeded, + timeout=timeout, + idle=self._idle, + cancelled=self._cancelled, + ) self._finalize_task(False, message, elapsed, max_turns_exceeded=max_turns_exceeded, timeout=timeout) + def _persist_conversation(self, *, partial_summary: str = "") -> None: + """每轮结束后立即落盘子智能体对话，避免跑完了才存一次导致中间状态丢失。""" + try: + runtime_seconds = int((time.time() * 1000 - self.stats["runtime_start"]) / 1000) + status = "running" + if self._cancelled: + status = "terminated" + elif self.multi_agent_mode and self._idle: + status = "idle" + ma_debug( + "sub_agent_persist_conversation", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + status=status, + idle=self._idle, + cancelled=self._cancelled, + multi_agent_mode=self.multi_agent_mode, + ) + conversation_data = { + "agent_id": self.agent_id, + "task_id": self.task_id, + "created_at": datetime.fromtimestamp(self.stats["runtime_start"] / 1000).isoformat(), + "updated_at": datetime.now().isoformat(), + "status": status, + "success": None, + "summary": partial_summary, + "messages": self.messages, + "stats": {**self.stats, "runtime_seconds": runtime_seconds, "turn_count": self.stats.get("turn_count", 0)}, + } + self.conversation_file.parent.mkdir(parents=True, exist_ok=True) + self.conversation_file.write_text(json.dumps(conversation_data, ensure_ascii=False), encoding="utf-8") + + output_data = { + "success": None, + "status": status, + "summary": partial_summary, + "stats": conversation_data["stats"], + } + self.output_file.parent.mkdir(parents=True, exist_ok=True) + self.output_file.write_text(json.dumps(output_data, ensure_ascii=False), encoding="utf-8") + except Exception as exc: + logger.warning(f"[SubAgentTask] 增量保存失败: {exc}") + def _finalize_task(self, success: bool, summary: str, elapsed: float, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None: runtime_seconds = int(elapsed) + ma_debug( + "sub_agent_finalize_task", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + success=success, + summary=summary, + idle=self._idle, + cancelled=self._cancelled, + ) output_data = { "success": success, "summary": summary, diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 6dd0395..c8b105d 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -826,13 +826,13 @@ def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: 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: + # 多智能体模式：只要有未消费的主对话消息，或还有未结束（非 idle/terminated）的实例，就继续轮询。 + # 即使全部实例都已 idle，只要 pending 消息尚未消费，也要继续，避免主任务结束后通知丢失。 + if getattr(web_terminal, "multi_agent_mode", False): 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 @@ -1839,6 +1839,7 @@ async def handle_task_with_sender( # 检查是否有后台运行的子智能体或待通知的完成任务 manager = getattr(web_terminal, "sub_agent_manager", None) has_running_sub_agents = False + has_running_multi_agent = False bg_manager = getattr(web_terminal, "background_command_manager", None) has_running_background_commands = False if manager: @@ -1848,16 +1849,33 @@ async def handle_task_with_sender( debug_log(f"[SubAgent] reconcile_task_states failed: {exc}") if not hasattr(web_terminal, "_announced_sub_agent_tasks"): web_terminal._announced_sub_agent_tasks = set() + + # 多智能体模式：检测是否还有运行中/idle 的多智能体实例。 + # 它们不是传统后台任务，不阻塞前端输入区，但需要通知池在主任务结束后 + # 继续消费 pending 消息并触发 Team Leader 响应。 + if getattr(web_terminal, "multi_agent_mode", False): + for task in manager.tasks.values(): + if ( + isinstance(task, dict) + and task.get("conversation_id") == conversation_id + and task.get("multi_agent_mode") + and task.get("status") not in TERMINAL_STATUSES.union({"terminated"}) + ): + has_running_multi_agent = True + break + running_tasks = [ task for task in manager.tasks.values() if task.get("status") not in TERMINAL_STATUSES.union({"terminated"}) and task.get("run_in_background") + and not task.get("multi_agent_mode") and task.get("conversation_id") == conversation_id ] pending_notice_tasks = [ task for task in manager.tasks.values() if task.get("status") in TERMINAL_STATUSES.union({"terminated"}) and task.get("run_in_background") + and not task.get("multi_agent_mode") and task.get("conversation_id") == conversation_id and task.get("task_id") not in web_terminal._announced_sub_agent_tasks ] @@ -1884,11 +1902,11 @@ async def handle_task_with_sender( # 与子智能体完全复用同一 waiting 事件（前端已有稳定处理链路） sender('sub_agent_waiting', _build_shared_waiting_payload(waiting_items)) - # 统一「通知池」轮询器：子智能体 + 后台 run_command 合并为单一轮询链路， - # 每轮一次性取出两路所有待通知项，避免逐条触发「工作 → 停止 → 再工作」循环。 + # 统一「通知池」轮询器：子智能体 + 后台 run_command + 多智能体 pending 消息 + # 合并为单一轮询链路，每轮一次性取出所有待通知项，避免逐条触发「工作 → 停止 → 再工作」循环。 # 只 spawn 一个轮询器（无论是否同时存在子智能体与后台命令），从根本上消除 # 两个轮询器同时 create_chat_task 撞「单工作区互斥」的问题。 - if has_running_sub_agents or has_running_background_commands: + if has_running_sub_agents or has_running_background_commands or has_running_multi_agent: def run_completion_poll(): import asyncio loop = asyncio.new_event_loop() diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index 48ff7d8..b66ffca 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -6,6 +6,7 @@ from datetime import datetime from typing import Any, Callable, Dict, List, Optional from modules.sub_agent import TERMINAL_STATUSES +from modules.multi_agent.debug_logger import ma_debug _VALID_SOURCES = { @@ -329,6 +330,14 @@ def inject_multi_agent_master_message( if not raw: return None + ma_debug( + "inject_multi_agent_master_message", + conversation_id=conversation_id, + text_preview=raw[:300], + inline=inline, + after_tool_call_id=after_tool_call_id, + ) + metadata = { "runtime_injected": True, "source": "sub_agent", @@ -405,6 +414,12 @@ async def process_multi_agent_master_messages( if not pending: return 0 debug_log(f"[MultiAgent] draining {len(pending)} pending master messages") + ma_debug( + "process_multi_agent_master_messages", + conversation_id=conversation_id, + count=len(pending), + previews=[str(m)[:200] for m in pending], + ) for msg in pending: inject_multi_agent_master_message( web_terminal=web_terminal, diff --git a/server/conversation.py b/server/conversation.py index 3c4db1c..7cdd84e 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -598,6 +598,7 @@ def create_conversation(terminal: WebTerminal, workspace: UserWorkspace, usernam thinking_mode = data.get('thinking_mode') if preserve_mode and 'thinking_mode' in data else None run_mode = data.get('mode') if preserve_mode and 'mode' in data else None target_workspace_id = (data.get('workspace_id') or '').strip() + multi_agent_mode = bool(data.get('multi_agent_mode')) if target_workspace_id: try: @@ -641,6 +642,7 @@ def create_conversation(terminal: WebTerminal, workspace: UserWorkspace, usernam metadata_overrides={ "permission_mode": default_permission_mode or getattr(terminal, "get_permission_mode", lambda: "unrestricted")(), "execution_mode": getattr(terminal, "get_execution_mode", lambda: "sandbox")(), + "multi_agent_mode": multi_agent_mode, }, ) try: @@ -654,7 +656,11 @@ def create_conversation(terminal: WebTerminal, workspace: UserWorkspace, usernam "safe_navigation": True, } else: - result = terminal.create_new_conversation(thinking_mode=thinking_mode, run_mode=run_mode) + result = terminal.create_new_conversation( + thinking_mode=thinking_mode, + run_mode=run_mode, + metadata_overrides={"multi_agent_mode": multi_agent_mode} if multi_agent_mode else None, + ) if result["success"]: # 仅在当前工作区创建时更新 session 模式；指定其他工作区时由前端切换后自动同步。 diff --git a/static/src/app/methods/conversation/action.ts b/static/src/app/methods/conversation/action.ts index 5b58af8..386521f 100644 --- a/static/src/app/methods/conversation/action.ts +++ b/static/src/app/methods/conversation/action.ts @@ -122,15 +122,17 @@ export const actionMethods = { this.versioningInitializingBackupToastId = backupToastId; } - const response = await fetch('/api/conversations', { + const isMultiAgent = Boolean(this.multiAgentMode); + const createUrl = isMultiAgent ? '/api/multiagent/conversations' : '/api/conversations'; + const createBody = isMultiAgent + ? JSON.stringify({ preserve_mode: true, thinking_mode: this.thinkingMode, mode: this.runMode }) + : JSON.stringify({ thinking_mode: this.thinkingMode, mode: this.runMode }); + const response = await fetch(createUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - thinking_mode: this.thinkingMode, - mode: this.runMode - }) + body: createBody }); const result = await response.json(); diff --git a/static/src/app/methods/message/send.ts b/static/src/app/methods/message/send.ts index c065805..113b97d 100644 --- a/static/src/app/methods/message/send.ts +++ b/static/src/app/methods/message/send.ts @@ -209,15 +209,17 @@ export const sendMethods = { this.versioningInitializingBackupToastId = backupToastId; } - const createResp = await fetch('/api/conversations', { + const isMultiAgent = Boolean(this.multiAgentMode); + const createUrl = isMultiAgent ? '/api/multiagent/conversations' : '/api/conversations'; + const createBody = isMultiAgent + ? JSON.stringify({ preserve_mode: true, thinking_mode: this.thinkingMode, mode: this.runMode }) + : JSON.stringify({ thinking_mode: this.thinkingMode, mode: this.runMode }); + const createResp = await fetch(createUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - thinking_mode: this.thinkingMode, - mode: this.runMode - }) + body: createBody }); const createResult = await createResp.json().catch(() => ({})); if (!createResp.ok || !createResult?.success || !createResult?.conversation_id) { @@ -258,7 +260,8 @@ export const sendMethods = { } const pathFragment = this.stripConversationPrefix(targetConversationId); - history.replaceState({ conversationId: targetConversationId }, '', `/${pathFragment}`); + const urlPrefix = this.multiAgentMode ? '/multiagent/' : '/'; + history.replaceState({ conversationId: targetConversationId }, '', `${urlPrefix}${pathFragment}`); } catch (error) { this.uiPushToast({ title: '发送失败', diff --git a/static/src/components/overlay/SubAgentActivityDialog.vue b/static/src/components/overlay/SubAgentActivityDialog.vue index 7c3dfcb..dc36665 100644 --- a/static/src/components/overlay/SubAgentActivityDialog.vue +++ b/static/src/components/overlay/SubAgentActivityDialog.vue @@ -29,13 +29,29 @@
{{ activityError }}
-
+
{{ activityLoading ? '正在读取子智能体活动...' : '暂无活动记录' }}
-
- {{ item.text }} - {{ item.stateLabel }} +
+ +
@@ -50,17 +66,22 @@ import { storeToRefs } from 'pinia'; import { useSubAgentStore } from '@/stores/subAgent'; type ActivityEntry = { + type?: string; id?: string; tool?: string; status?: string; args?: Record; ts?: number; + subtype?: string; + content?: string; + is_final?: boolean; }; const subAgentStore = useSubAgentStore(); const { activeAgent, activityEntries, activityLoading, activityError, stoppingTaskIds } = storeToRefs(subAgentStore); const stopError = ref(''); +const expandedOutputs = ref>(new Set()); const close = () => { subAgentStore.closeSubAgent(); @@ -141,48 +162,138 @@ const isTerminalStatus = (status?: string) => { return ['completed', 'failed', 'timeout', 'terminated', 'cancelled'].includes(normalized); }; -const displayItems = computed(() => { +const toggleOutput = (key: string) => { + const next = new Set(expandedOutputs.value); + if (next.has(key)) { + next.delete(key); + } else { + next.add(key); + } + expandedOutputs.value = next; +}; + +const handleItemClick = (item: any) => { + if (item.kind === 'output') { + toggleOutput(item.key); + } +}; + +const timelineItems = computed(() => { const entries = activityEntries.value || []; - const groups: { key: string; entry: ActivityEntry }[] = []; + const rawItems: { kind: 'tool'; key: string; entry: ActivityEntry } | { kind: 'output'; key: string; content: string; isFinal: boolean }[] = []; + let currentToolGroup: { kind: 'tool'; key: string; entry: ActivityEntry } | null = null; - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]; - if (!entry || entry.type !== 'progress') continue; + const flushToolGroup = () => { + if (currentToolGroup) { + rawItems.push(currentToolGroup); + currentToolGroup = null; + } + }; - const baseKey = entry.id || `${entry.tool || 'tool'}-${entry.ts || i}`; - const lastGroup = groups[groups.length - 1]; - - // 同一个 id 的相邻进度更新（running -> completed）合并为一次工具调用。 - // 如果上一个同 id 的组已经进入终态，则后续同 id 的记录视为新的工具调用。 - if ( - lastGroup && - (lastGroup.entry.id === entry.id || lastGroup.key === baseKey) && - !isTerminalStatus(lastGroup.entry.status) - ) { - lastGroup.entry = { ...lastGroup.entry, ...entry }; - continue; + entries.forEach((entry: ActivityEntry, index: number) => { + if (entry?.type === 'progress' && entry?.subtype === 'output' && typeof entry.content === 'string') { + flushToolGroup(); + rawItems.push({ + kind: 'output', + key: `output-${entry.ts || index}`, + content: entry.content, + isFinal: !!entry.is_final, + }); + return; } - // 生成唯一 key：id 重复时加上序号 + if (!entry || entry.type !== 'progress' || !entry.tool) return; + + const baseKey = entry.id || `${entry.tool}-${entry.ts || index}`; + if ( + currentToolGroup && + (currentToolGroup.entry.id === entry.id || currentToolGroup.key === baseKey) && + !isTerminalStatus(currentToolGroup.entry.status) + ) { + currentToolGroup.entry = { ...currentToolGroup.entry, ...entry }; + return; + } + + flushToolGroup(); let key = baseKey; let suffix = 0; - while (groups.some((g) => g.key === key)) { + while (rawItems.some((item) => item.kind === 'tool' && item.key === key)) { suffix++; key = `${baseKey}--${suffix}`; } - groups.push({ key, entry: { ...entry } }); - } + currentToolGroup = { kind: 'tool', key, entry: { ...entry } }; + }); - return groups.map((group) => { - const item = group.entry; - const state = normalizeStatus(item.status); - const stateLabel = state === 'completed' ? '完成' : state === 'failed' ? '失败' : '进行中'; + flushToolGroup(); + + return rawItems.map((item) => { + if (item.kind === 'output') return item; + const state = normalizeStatus(item.entry.status); return { - key: group.key, + kind: 'tool' as const, + key: item.key, state, - stateLabel, - text: buildText(item) + stateLabel: state === 'completed' ? '完成' : state === 'failed' ? '失败' : '进行中', + text: buildText(item.entry) }; }); }); + + diff --git a/static/src/composables/useLegacySocket.ts b/static/src/composables/useLegacySocket.ts index 435796c..a001294 100644 --- a/static/src/composables/useLegacySocket.ts +++ b/static/src/composables/useLegacySocket.ts @@ -1590,7 +1590,8 @@ export async function initializeLegacySocket(ctx: any) { const hasRunningSubAgents = !!data?.has_running_sub_agents; const hasRunningBackgroundCommands = !!data?.has_running_background_commands; const hasRunningBackground = hasRunningSubAgents || hasRunningBackgroundCommands; - if (!hasRunningBackground) { + // 多智能体模式下，主任务停止即可恢复输入区，不因子智能体后台运行保持停止按钮 + if (ctx.multiAgentMode || !hasRunningBackground) { ctx.taskInProgress = false; } else { ctx.waitingForSubAgent = hasRunningSubAgents; @@ -1616,8 +1617,8 @@ export async function initializeLegacySocket(ctx: any) { }); socketLog('任务完成', data); - // 如果有运行中的子智能体，不重置任务状态 - if (!data.has_running_sub_agents) { + // 多智能体模式下，主任务完成即视为可发送状态，后台子智能体是否 idle/running 不影响输入区 + if (ctx.multiAgentMode || !data.has_running_sub_agents) { console.log('[DEBUG] 没有运行中的子智能体，重置任务状态'); if (ctx.waitingForSubAgent) { ctx.waitingForSubAgent = false; @@ -1652,13 +1653,16 @@ export async function initializeLegacySocket(ctx: any) { }); socketLog('等待子智能体完成:', data); - // 设置标志：有子智能体在运行，阻止状态重置 - ctx.waitingForSubAgent = true; + // 多智能体模式下，子智能体 idle/running 是常态，不阻塞主对话输入区 + if (!ctx.multiAgentMode) { + // 设置标志：有子智能体在运行，阻止状态重置 + ctx.waitingForSubAgent = true; - // 保持任务进行中状态 - ctx.taskInProgress = true; - ctx.streamingMessage = false; - ctx.stopRequested = false; + // 保持任务进行中状态 + ctx.taskInProgress = true; + ctx.streamingMessage = false; + ctx.stopRequested = false; + } console.log('[DEBUG] 当前状态 (after sub_agent_waiting):', { taskInProgress: ctx.taskInProgress,