From a1a9a474464733ed6b4a43ead325d8bf46c3ff97 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Tue, 21 Jul 2026 15:46:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(multi-agent):=20=E7=BB=88=E7=BB=93=E5=AD=90?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E6=94=B9=E4=B8=BA=E5=90=B8=E6=94=B6?= =?UTF-8?q?=E6=80=81=EF=BC=8C=E4=BF=AE=E5=A4=8D=E7=BB=88=E7=BB=93=E5=90=8E?= =?UTF-8?q?=E5=A4=8D=E6=B4=BB/=E5=81=9C=E6=AD=A2=E6=97=A0=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit terminate_sub_agent 彻底化： - 通过 _sub_agent_instances 找到活实例直接 cancel 其 _task（原实现只查 _running_tasks，多 manager 并存时拿不到句柄导致 cancel 未送达） - 实例置 _cancelled 并从注册表移除，防止 inject 找到活实例直接复活 - MultiAgentState 同步标记 terminated（原实现不写，侧边栏一直显示 idle） - output.json 改写为 terminated 终态快照（原实现保留陈旧 idle 快照， 会被 _check_task_status 读回复活任务记录） - 新增 control.json 跨 manager 击杀/软停止信号通道，子智能体运行循环 每个 idle tick / 每轮开始自行消费 注入/复活路径拒绝终结者： - inject_message_to_sub_agent 检查最新任务记录，terminated 拒绝注入并 清理孤儿实例；_revive_sub_agent 排除 terminated 记录 - send_message_to_sub_agent 工具返回明确的'已被终结'错误 - stop_sub_agent / soft_stop_all_agents 跳过 terminated，本地无实例时 经 control.json 让对端 manager 上的实例自行软停止 terminated 吸收态防护： - _check_task_status/_handle_running_snapshot 不再用陈旧快照覆盖终结记录 - output.json 为 terminated 时同步本地记录（跨 manager 终结广播） - reconcile_task_states 的 idle/running 修正跳过终态任务 - MultiAgentState.mark_status 禁止 terminated 被覆盖回 idle/running - 硬取消路径 output.json 写 terminated 而非 failed，避免 failed 被复活 --- core/main_terminal_parts/tools_execution.py | 6 +- modules/multi_agent/state.py | 9 + modules/sub_agent/creation.py | 14 +- modules/sub_agent/manager.py | 206 ++++++++++++++++++-- modules/sub_agent/state.py | 10 + modules/sub_agent/task.py | 76 +++++++- 6 files changed, 298 insertions(+), 23 deletions(-) diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index dc0d4bb4..e4e64f77 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -1951,7 +1951,11 @@ class MainTerminalToolsExecutionMixin: ) ok = self.sub_agent_manager.inject_message_to_sub_agent(agent_id, text) if not ok: - result = {"success": False, "error": f"子智能体 {agent_id} 不存在或已结束"} + latest = self.sub_agent_manager._latest_task_for_agent(agent_id) + if latest and latest.get("status") == "terminated": + result = {"success": False, "error": f"子智能体 {agent_id} 已被手动终结，无法再接收消息。如需继续工作，请创建新的子智能体。"} + else: + result = {"success": False, "error": f"子智能体 {agent_id} 不存在或已结束"} else: result = {"success": True, "agent_id": agent_id} ma_debug( diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index 72c743cc..6d2793c8 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -326,6 +326,15 @@ class MultiAgentState: a = self.agents.get(agent_id) if not a: return + # 多智能体原则：terminated 是真正终结（吸收态），不允许被覆盖回 idle/running + if a.status == "terminated" and status != "terminated": + ma_debug( + "mark_status_blocked_terminal", + agent_id=agent_id, + current_status=a.status, + requested_status=status, + ) + return a.status = status if last_output: a.last_output = last_output diff --git a/modules/sub_agent/creation.py b/modules/sub_agent/creation.py index 5abad99c..46a16148 100644 --- a/modules/sub_agent/creation.py +++ b/modules/sub_agent/creation.py @@ -17,15 +17,25 @@ class SubAgentCreationMixin: conversation_agents: Dict[str, List[int]] project_path: Path - def _select_task(self, task_id: Optional[str], agent_id: Optional[int]) -> Optional[Dict]: + def _select_task( + self, + task_id: Optional[str], + agent_id: Optional[int], + *, + include_idle: bool = False, + ) -> Optional[Dict]: self.reconcile_task_states() if task_id: return self.tasks.get(task_id) if agent_id is None: return None + # 多智能体模式下子智能体常驻 idle（等待唤醒），terminate/stop 需要能选中它们 + allowed = {"pending", "running"} + if include_idle: + allowed = allowed | {"idle"} candidates = [ task for task in self.tasks.values() - if task.get("agent_id") == agent_id and task.get("status") in {"pending", "running"} + if task.get("agent_id") == agent_id and task.get("status") in allowed ] if candidates: candidates.sort(key=lambda item: item.get("created_at", 0), reverse=True) diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index 46a36ae8..b236def2 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -435,6 +435,9 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi except Exception as exc: ma_debug("soft_stop_failed", task_id=task_id, error=str(exc)) else: + # 实例跑在另一个 manager 内存里：写控制文件让其自行软停止 + self._write_sub_agent_control_request(task_info, "soft_stop") + count += 1 skipped_no_instance += 1 ma_debug( "soft_stop_all_agents_done", @@ -453,7 +456,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi agent_id: Optional[int] = None, ) -> Dict: """暂停指定子智能体，使其进入 idle 状态而不终结。""" - task = self._select_task(task_id, agent_id) + task = self._select_task(task_id, agent_id, include_idle=True) if not task: return {"success": False, "error": "未找到对应的子智能体任务"} @@ -461,6 +464,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi real_agent_id = task.get("agent_id") if not task.get("multi_agent_mode"): return {"success": False, "error": "stop_sub_agent 仅在多智能体模式下可用"} + if task.get("status") == "terminated": + return {"success": False, "error": "子智能体已被终结，无法暂停"} # 查找或复活实例，确保能接收软停止信号 if real_agent_id is not None: @@ -474,7 +479,9 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi except Exception as exc: return {"success": False, "error": f"暂停子智能体失败: {exc}"} else: - # 没有活实例时直接修改任务状态为 idle + # 本地没有活实例：实例可能跑在另一个 manager 内存里， + # 写控制文件让对端自行软停止；同时先把记录置 idle 作为即时反馈 + self._write_sub_agent_control_request(task, "soft_stop") task["status"] = "idle" task["updated_at"] = time.time() self._save_state() @@ -505,39 +512,180 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi task_id: Optional[str] = None, agent_id: Optional[int] = None, ) -> Dict: - """强制关闭指定子智能体。""" - task = self._select_task(task_id, agent_id) + """强制关闭指定子智能体。 + + terminated 是吸收态，必须同时清理四层状态，否则会被复活： + 1) 内存实例（_sub_agent_instances / _running_tasks）——取消并移除； + 2) 任务记录 status —— 标记 terminated； + 3) MultiAgentState 实例状态 —— 标记 terminated； + 4) output.json 快照 —— 改写为 terminated，防止陈旧 idle 快照被读回复活； + 另外写 control.json 作为跨 manager 击杀通道（实例可能跑在别的 manager 内存里）。 + """ + # 多智能体模式下子智能体可能处于 idle，仍需支持终结 + task = self._select_task(task_id, agent_id, include_idle=True) if not task: return {"success": False, "error": "未找到对应的子智能体任务"} 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: - time.sleep(0.05) + agent_id = task.get("agent_id") + display_name = task.get("display_name") or f"子智能体{agent_id}" + ma_debug( + "terminate_sub_agent_enter", + task_id=task_id, + agent_id=agent_id, + before_status=task.get("status"), + multi_agent_mode=task.get("multi_agent_mode"), + ) + # ---- 1) 取消内存中的 asyncio 任务 ---- + # _running_tasks 可能因多 manager 并存/注册缺失而拿不到真正的句柄， + # 必须同时通过 _sub_agent_instances 找到活实例直接取消它的 _task。 + inst = self._sub_agent_instances.get(agent_id) if agent_id is not None else None + cancel_targets: List[Any] = [] + if inst is not None: + # cancel 未送达时的兜底：让 run 循环在下一个 tick 自行退出 + inst._cancelled = True + inst_task = getattr(inst, "_task", None) + if inst_task is not None: + cancel_targets.append(inst_task) + running_task = self._running_tasks.pop(task_id, None) + if running_task is not None and running_task not in cancel_targets: + cancel_targets.append(running_task) + for target in cancel_targets: + if target.done(): + continue + try: + # 子智能体运行在独立事件循环线程中，取消操作必须投递到该循环 + loop = target.get_loop() + loop.call_soon_threadsafe(target.cancel) + except Exception: + try: + target.cancel() + except Exception: + pass + if cancel_targets: + deadline = time.time() + 5 + for target in cancel_targets: + while not target.done() and time.time() < deadline: + time.sleep(0.05) + + # 实例从注册表移除，防止后续 inject/revive 找到它直接复活 + if agent_id is not None: + self._sub_agent_instances.pop(agent_id, None) + + # ---- 2) 任务记录标记 terminated（含实例对应记录，若与选中记录不同） ---- self._mark_task_terminated( task, message="子智能体已被强制关闭。", - system_message=f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。", + system_message=f"🛑 {display_name} 已被手动关闭。", notified=True, ) + inst_task_id = getattr(inst, "task_id", None) if inst is not None else None + if inst_task_id and inst_task_id != task_id: + other_task = self.tasks.get(inst_task_id) + if other_task and other_task.get("status") not in TERMINAL_STATUSES.union({"terminated"}): + self._mark_task_terminated( + other_task, + message="子智能体已被强制关闭。", + system_message=f"🛑 {display_name} 已被手动关闭。", + notified=True, + ) + + # ---- 3) MultiAgentState 实例状态标记 terminated ---- + conversation_id = task.get("conversation_id") + if conversation_id and agent_id is not None: + state = self.get_multi_agent_state(conversation_id) + if state: + state.mark_status(agent_id, "terminated") + + # ---- 4) output.json 终态快照 + control.json 跨 manager 击杀信号 ---- + self._write_terminated_output_snapshot(task, display_name=display_name) + self._write_sub_agent_control_request(task, "terminate") + self._save_state() + ma_debug( + "terminate_sub_agent_done", + task_id=task_id, + agent_id=agent_id, + had_instance=bool(inst), + cancel_targets=len(cancel_targets), + ) return { "success": True, "task_id": task_id, + "agent_id": agent_id, + "display_name": display_name, "message": "子智能体已被强制关闭。", - "system_message": f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。", + "system_message": f"🛑 {display_name} 已被手动关闭。", } + def _write_terminated_output_snapshot(self, task: Dict, *, display_name: str = "") -> None: + """把 output.json 改写为 terminated 终态快照。 + + 防止子智能体此前周期写入的 {status: idle/running, success: null} 陈旧快照 + 被 _check_task_status 读回后把任务记录复活为 idle/running；同时作为跨 + manager 的终结广播：其他 manager 读到 terminated 快照后同步本地记录。 + """ + try: + raw_output_file = task.get("output_file", "") + if not raw_output_file: + return + output_file = Path(raw_output_file) + try: + existing = json.loads(output_file.read_text(encoding="utf-8")) if output_file.exists() else {} + except Exception: + existing = {} + if not isinstance(existing, dict): + existing = {} + existing.update({ + "success": False, + "status": "terminated", + "summary": f"{display_name or '子智能体'} 已被手动关闭。", + "terminated_at": time.time(), + }) + output_file.parent.mkdir(parents=True, exist_ok=True) + output_file.write_text(json.dumps(existing, ensure_ascii=False), encoding="utf-8") + except Exception as exc: + logger.warning(f"[terminate] 写入终态快照失败: {exc}") + + def _write_sub_agent_control_request(self, task: Dict, action: str) -> None: + """向子智能体任务目录写控制请求（跨 manager 信号通道）。 + + 子智能体的 asyncio 实例可能跑在另一个 manager 的内存里（多 WebTerminal + 并存），本 manager 无法直接 cancel。子智能体运行循环在每个 idle tick / + 每轮开始都会读取 control.json，看到请求后自行执行 terminate/soft_stop。 + """ + try: + raw_output_file = task.get("output_file", "") + if not raw_output_file: + return + control_file = Path(raw_output_file).parent / "control.json" + control_file.parent.mkdir(parents=True, exist_ok=True) + payload = {"action": action, "requested_at": time.time()} + tmp_file = control_file.with_suffix(".tmp") + tmp_file.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + tmp_file.replace(control_file) + except Exception as exc: + logger.warning(f"[control] 写入控制请求失败: {exc}") + + def _latest_task_for_agent(self, agent_id: int, conversation_id: Optional[str] = None) -> Optional[Dict]: + """返回指定 agent_id 最新一条任务记录（不限状态）。 + + agent_id 只在单会话内唯一，tasks 字典跨会话共享，必须按会话过滤， + 否则其他会话的同名 agent 状态会污染判断（如误拒注入）。 + """ + cid = conversation_id or getattr(self, "owner_conversation_id", None) + candidates = [ + t for t in self.tasks.values() + if isinstance(t, dict) and t.get("agent_id") == agent_id + and (not cid or t.get("conversation_id") == cid) + ] + if not candidates: + return None + candidates.sort(key=lambda item: item.get("created_at", 0), reverse=True) + return candidates[0] + def get_sub_agent_status( self, *, @@ -944,6 +1092,9 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi continue if not task.get("multi_agent_mode"): continue + # terminated 是吸收态：idle/running 修正不得触碰终结任务 + if task.get("status") in TERMINAL_STATUSES.union({"terminated"}): + continue agent_id = task.get("agent_id") inst = self._sub_agent_instances.get(agent_id) if agent_id else None if inst: @@ -981,6 +1132,26 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi 若内存中无运行实例（如 failed 后保留的实例已结束），会尝试从 conversation 文件重建子智能体（保留原 agent_id 和 role_id）后再注入消息。 """ + # 终结检查：该 agent 最新任务记录为 terminated 时拒绝注入， + # 并顺带清理可能残活的内存实例（孤儿实例），防止复活 + latest_task = self._latest_task_for_agent(agent_id) + if latest_task and latest_task.get("status") == "terminated": + orphan = self._sub_agent_instances.pop(agent_id, None) + if orphan is not None: + orphan._cancelled = True + orphan_task = getattr(orphan, "_task", None) + if orphan_task is not None and not orphan_task.done(): + try: + orphan_task.get_loop().call_soon_threadsafe(orphan_task.cancel) + except Exception: + pass + ma_debug( + "inject_message_rejected_terminated", + agent_id=agent_id, + task_id=latest_task.get("task_id"), + had_orphan_instance=bool(orphan), + ) + return False # 查找或复活该 agent_id 对应的 SubAgentTask sub_agent = self._find_or_revive_sub_agent_task(agent_id) ma_debug( @@ -1017,6 +1188,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi candidates = [ t for t in self.tasks.values() if isinstance(t, dict) and t.get("agent_id") == agent_id and t.get("multi_agent_mode") + and t.get("status") != "terminated" # terminated 是吸收态，不可复活 ] if not candidates: return None diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index f3527972..0f52a6e7 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -129,6 +129,9 @@ class SubAgentStateMixin: def _check_task_status(self, task: Dict) -> Dict: """检查任务状态，读取输出文件。""" task_id = task["task_id"] + # terminated 是吸收态：不允许 output.json 中的陈旧 running/idle 快照复活 + if task.get("status") == "terminated": + return {"success": False, "status": "terminated", "task_id": task_id} output_file = Path(task.get("output_file", "")) if not output_file.exists(): running_task = self._running_tasks.get(task_id) @@ -155,6 +158,13 @@ class SubAgentStateMixin: stats = output.get("stats", {}) elapsed_seconds = self._compute_elapsed_seconds(task) + # output.json 被写入终态快照（terminate 的跨 manager 击杀通道）：同步记录为终结 + if output.get("status") == "terminated": + task["status"] = "terminated" + task["updated_at"] = time.time() + ma_debug("check_task_status_output_terminated", task_id=task_id, agent_id=task.get("agent_id")) + return {"success": False, "status": "terminated", "task_id": task_id, "message": summary or "子智能体已被终结"} + ma_debug( "check_task_status", task_id=task_id, diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index f67a2072..e20442c6 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -158,15 +158,18 @@ class SubAgentTask: except asyncio.CancelledError: # 二次取消视为硬取消 self._cancelled = True - await asyncio.shield(self._write_failure("子智能体被手动终止")) + self._persist_conversation(partial_summary="子智能体被手动终止") + self._write_terminated_snapshot() raise return # 硬取消（terminate）路径 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("子智能体被手动终止")) + # 写 terminated 终态快照而不是 failed：terminated 是吸收态， + # 避免 output.json 的 failed 被其他 manager 读回后走复活路径 + self._persist_conversation(partial_summary="子智能体被手动终止") + self._write_terminated_snapshot() raise except Exception as exc: logger.exception(f"[SubAgent] task={self.task_id} 执行异常") @@ -193,6 +196,31 @@ class SubAgentTask: await self._write_timeout(elapsed) return + # 跨 manager 控制信号：terminate/soft_stop 可能由另一个 manager 经 + # control.json 下达（其实例无法直接操作本实例），每个 tick 检查一次 + control_action = self._consume_control_request() + if control_action == "terminate": + self._cancelled = True + ma_debug( + "sub_agent_control_terminate", + 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, "terminated") + self._persist_conversation(partial_summary="子智能体已被手动终止") + self._write_terminated_snapshot() + return + if control_action == "soft_stop" and not self._idle: + ma_debug( + "sub_agent_control_soft_stop", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + ) + self._soft_stop = True + # 多智能体模式下，idle 时等待新消息或外部回答；只有真正被注入消息时才继续运行 if self.multi_agent_mode and self._idle: event_set = False @@ -1089,6 +1117,48 @@ class SubAgentTask: self.manager._mark_task_done(self.task_id, success, summary, runtime_seconds) + def _consume_control_request(self) -> Optional[str]: + """读取并消费 control.json 控制请求（跨 manager 信号通道）。 + + terminate_sub_agent / stop_sub_agent 可能运行在另一个 manager 的内存里， + 无法直接操作本实例，通过任务目录下的 control.json 下达指令。 + 返回 "terminate" / "soft_stop" / None。 + """ + control_file = self.output_file.parent / "control.json" + try: + if not control_file.exists(): + return None + try: + data = json.loads(control_file.read_text(encoding="utf-8")) + except Exception: + data = {} + action = str(data.get("action") or "").strip() if isinstance(data, dict) else "" + # 消费后删除，避免重复执行；删除失败不阻塞主流程 + try: + control_file.unlink() + except Exception: + pass + if action in {"terminate", "soft_stop"}: + return action + except Exception: + pass + return None + + def _write_terminated_snapshot(self) -> None: + """写入 terminated 终态快照到 output.json（硬取消/自杀路径）。""" + try: + runtime_seconds = int(time.time() - (self.stats["runtime_start"] / 1000)) + output_data = { + "success": False, + "status": "terminated", + "summary": "子智能体已被手动终止", + "stats": {**self.stats, "runtime_seconds": runtime_seconds}, + } + 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 cancel(self) -> None: self._cancelled = True if self._task and not self._task.done():