From 77a271fa23f260771cced7dd55f1993d29ce24ae Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 22 Jul 2026 00:53:05 +0800 Subject: [PATCH] =?UTF-8?q?fix(multi-agent):=20MultiAgentState=20=E5=85=A8?= =?UTF-8?q?=E5=B1=80=E5=8D=95=E4=BE=8B=EF=BC=8C=E6=A0=B9=E6=B2=BB=20termin?= =?UTF-8?q?ate=20=E5=90=8E=E7=8A=B6=E6=80=81=E5=88=86=E8=A3=82=E5=A4=8D?= =?UTF-8?q?=E6=B4=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MultiAgentState 从 manager 实例属性改为进程级全局注册表 GLOBAL_MULTI_AGENT_STATES （多 manager 并存时各 from_snapshot 出独立副本，terminate 只标记一份，轮询落到陈旧副本显示空闲） - _load_state 恢复后用任务记录校准实例终态（磁盘快照可能被陈旧副本洗回 idle） - 显示名后缀固定用角色内编号（role_seq 每次创建必递增），与内部 agent_id 命名空间分离； 存量 _None 后缀按对话×角色×创建时间自愈迁移 - tools_execution 同步含 edited_files 工具埋点（write/edit/delete/rename） --- AGENTS.md | 4 +- core/main_terminal_parts/tools_execution.py | 115 +++++++++++++++++++- modules/multi_agent/state.py | 19 +++- modules/sub_agent/manager.py | 38 ++++--- modules/sub_agent/state.py | 101 ++++++++++++++--- 5 files changed, 239 insertions(+), 38 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index fca03ee8..7714d901 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -466,7 +466,9 @@ AI 执行以下流程时，每一步都要向用户说明在做什么： - 切换会话不清理 `_running_tasks` 和 `_sub_agent_instances`。`SubAgentManager` 的全局 tasks 字典按 `task_id` 保持，多智能体状态由 `conversation_id` 在 `get_multi_agent_state` 中查。 - 子智能体对话存在 `~/.astrion/astrion/host/host/data/sub_agents/`。重启后走 `manager.restore_sub_agent` 恢复实例引用。 -- `MultiAgentState` 实例在 Manager 上常驻（dict by `conversation_id`）；进程重启走 `from_snapshot` 恢复。 +- **`MultiAgentState` 是进程级全局单例**（2026-07 重构）：存放在 `modules/multi_agent/state.py` 的 `GLOBAL_MULTI_AGENT_STATES`（key=`conversation_id`），所有 `SubAgentManager` 实例共享；`manager.multi_agent_states` 只是该全局 dict 的引用。此前它是 manager 实例属性，对话级 terminal 缓存重建会产生多个 manager，各自的 `_load_state` 都从磁盘快照 `from_snapshot` 出一份独立副本，导致 terminate 只标记其中一份、前端轮询落到其他副本显示陈旧 idle。`get_or_create` / `drop` / `_load_state` restore 均通过 `GLOBAL_MULTI_AGENT_STATES_LOCK`（RLock）互斥。 +- **进程重启后的状态校准**（2026-07 新增）：`_load_state` 恢复 ma 快照后，会用任务记录（持久真相）校准实例终态——任务记录是 terminated/终态而快照里还是 idle 的，一律校准为终态；同处还有存量 `_None` 后缀显示名的自愈迁移（按「对话×角色×创建时间」重编号）。 +- **显示名编号语义**：显示名后缀（如 `Full-Stack Engineer_1`）是**角色内编号**（`next_agent_id_for_role`，每次创建必递增），与内部 `agent_id`（LLM 可手动指定，如 10001）是两套独立命名空间，创建路径 `tools_execution.py` 中不得混用。 --- diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 605191d4..fd5cbd66 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -638,6 +638,103 @@ class MainTerminalToolsExecutionMixin: sorted(visited), ) + # ---------------- 快捷窗口：本次对话编辑/创建文件记录 ---------------- + + def _get_conversation_manager_for_edited_files(self, conversation_id: str): + cm = getattr(self, "context_manager", None) + if cm is None: + return None + router = getattr(cm, "_get_conversation_manager_for_id", None) + if callable(router): + try: + return router(conversation_id) + except Exception: + pass + return getattr(cm, "conversation_manager", None) + + def _mutate_edited_files(self, mutator) -> None: + """读取-修改-写回当前对话 metadata.edited_files，并实时广播给前端。""" + try: + cm = getattr(self, "context_manager", None) + if cm is None: + return + conv_id = getattr(self, "_bound_conversation_id", None) or getattr(cm, "current_conversation_id", None) + if not conv_id: + return + manager = self._get_conversation_manager_for_edited_files(conv_id) + if manager is None: + return + data = manager.load_conversation(conv_id) + if not data: + return + metadata = data.get("metadata", {}) or {} + raw = metadata.get("edited_files") + entries = [dict(item) for item in raw if isinstance(item, dict)] if isinstance(raw, list) else [] + changed = mutator(entries) + if not changed: + return + if not manager.update_conversation_metadata(conv_id, {"edited_files": entries}): + return + callback = getattr(cm, "_web_terminal_callback", None) + if callable(callback): + try: + callback("edited_files_updated", { + "conversation_id": conv_id, + "edited_files": entries, + }) + except Exception: + pass + except Exception as exc: + print(f"⚠️ 更新对话编辑文件记录失败: {exc}") + + def _record_edited_file(self, path: Any, op: str) -> None: + """记录本次对话中编辑/创建过的文件（相对路径，同 path 去重更新）。""" + rel_path = str(path or "").strip().replace("\\", "/") + if not rel_path: + return + + def _upsert(entries: List[Dict[str, Any]]) -> bool: + now = datetime.now().isoformat() + for item in entries: + if item.get("path") == rel_path: + item["op"] = op + item["ts"] = now + return True + entries.append({"path": rel_path, "op": op, "ts": now}) + return True + + self._mutate_edited_files(_upsert) + + def _remove_edited_file(self, path: Any) -> None: + """文件被删除后从记录中移除。""" + rel_path = str(path or "").strip().replace("\\", "/") + if not rel_path: + return + + def _remove(entries: List[Dict[str, Any]]) -> bool: + before = len(entries) + entries[:] = [item for item in entries if item.get("path") != rel_path] + return len(entries) != before + + self._mutate_edited_files(_remove) + + def _rename_edited_file(self, old_path: Any, new_path: Any) -> None: + """文件重命名后同步更新记录中的路径。""" + old_rel = str(old_path or "").strip().replace("\\", "/") + new_rel = str(new_path or "").strip().replace("\\", "/") + if not old_rel or not new_rel or old_rel == new_rel: + return + + def _rename(entries: List[Dict[str, Any]]) -> bool: + for item in entries: + if item.get("path") == old_rel: + item["path"] = new_rel + item["ts"] = datetime.now().isoformat() + return True + return False + + self._mutate_edited_files(_rename) + def _track_shallow_versioning(self, file_path: Any) -> None: """Track a file edit for shallow versioning (conversation-scoped undo).""" conversation_id = getattr(self.context_manager, "current_conversation_id", None) @@ -1297,6 +1394,8 @@ class MainTerminalToolsExecutionMixin: # 如果删除成功，同时删除备注 if result.get("success") and result.get("action") == "deleted": deleted_path = result.get("path") + # 快捷窗口：从编辑文件记录中移除 + self._remove_edited_file(deleted_path) # 删除备注 if deleted_path in self.context_manager.file_annotations: del self.context_manager.file_annotations[deleted_path] @@ -1313,6 +1412,8 @@ class MainTerminalToolsExecutionMixin: if result.get("success") and result.get("action") == "renamed": old_path = result.get("old_path") new_path = result.get("new_path") + # 快捷窗口：同步编辑文件记录中的路径 + self._rename_edited_file(old_path, new_path) # 更新备注 if old_path in self.context_manager.file_annotations: annotation = self.context_manager.file_annotations[old_path] @@ -1339,6 +1440,8 @@ class MainTerminalToolsExecutionMixin: # write_file 成功后，该文件视作当前会话已“接触”， # 允许后续继续 write/edit 而不再触发先读拦截。 self._mark_file_as_read_visited(result.get("path") or path) + # 快捷窗口：记录本次对话写入过的文件 + self._record_edited_file(result.get("path") or path, "write") elif tool_name == "edit_file": read_guard_error = self._check_read_before_edit_prerequisite(tool_name, arguments) @@ -1356,6 +1459,8 @@ class MainTerminalToolsExecutionMixin: result = self.file_manager.replace_many_in_file(path, replacements) if isinstance(result, dict) and result.get("success"): self._mark_file_as_read_visited(result.get("path") or path) + # 快捷窗口：记录本次对话编辑过的文件 + self._record_edited_file(result.get("path") or path, "edit") elif tool_name == "create_folder": result = self.file_manager.create_folder(arguments["path"]) @@ -1798,12 +1903,16 @@ class MainTerminalToolsExecutionMixin: else: conv_id = self.context_manager.current_conversation_id multi_agent_state = self.sub_agent_manager.get_or_create_multi_agent_state(conv_id) - # 分配 agent_id：如果未传入则自动递增 + # 角色内编号：每次创建都递增，作为显示名后缀 + # （如 Full-Stack Engineer_1）。它与内部 agent_id 是两套 + # 独立命名空间——agent_id 可被 LLM 手动指定（如 10001）， + # 显示名后缀永远用角色内编号。 + role_seq = multi_agent_state.next_agent_id_for_role(role_id) agent_id = arguments.get("agent_id") if not agent_id: - agent_id = multi_agent_state.next_agent_id_for_role(role_id) + agent_id = role_seq # 构造显示名 - display_name = role.display_name(int(agent_id)) + display_name = role.display_name(int(role_seq)) # 构造多智能体版系统提示词（含动态上下文注入） workspace_path = str(getattr(self, "project_path", "")) data_dir = str(getattr(self, "data_dir", "")) diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index 6d2793c8..b40cdb52 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -20,6 +20,7 @@ from __future__ import annotations import asyncio import json import re +import threading import uuid from asyncio import AbstractEventLoop from dataclasses import dataclass, field @@ -624,4 +625,20 @@ class MultiAgentState: self.pending_question_loops.clear() self.pending_answers.clear() self.agent_blocking_question.clear() - self.role_counters.clear() \ No newline at end of file + self.role_counters.clear() + + +# ---------------------------------------------------------------------- +# 进程级全局注册表（所有 SubAgentManager 共享） +# ---------------------------------------------------------------------- +# MultiAgentState 是会话级运行态，本质上不属于任何单个 SubAgentManager。 +# 历史上它存放在 manager.multi_agent_states（manager 实例属性）里，而对话级 +# terminal 缓存重建会产生多个 manager；每个 manager 的 _load_state 都从磁盘快照 +# from_snapshot 恢复一份副本，导致同一对话的 MultiAgentState 在内存中同时存在 +# N 份（terminate 只标记了其中一份，其余副本仍是陈旧 idle，前端轮询落到哪份 +# 就看到哪份的状态）。现在全进程共享同一份注册表，任何 manager 读写的都是同一 +# 对象，从根上消除多副本分裂。 +GLOBAL_MULTI_AGENT_STATES: Dict[str, "MultiAgentState"] = {} +# get-or-create / drop / _load_state restore 的 check-then-act 需要互斥； +# 用 RLock 防止与调用方已有锁重入死锁。 +GLOBAL_MULTI_AGENT_STATES_LOCK = threading.RLock() \ No newline at end of file diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index b236def2..a180c5a9 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -29,6 +29,10 @@ 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 +from modules.multi_agent.state import ( + GLOBAL_MULTI_AGENT_STATES, + GLOBAL_MULTI_AGENT_STATES_LOCK, +) from server.utils_common import debug_log if TYPE_CHECKING: @@ -63,9 +67,11 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi self.container_session: Optional["ContainerHandle"] = container_session self.host_execution_mode: str = "sandbox" self.terminal: Optional["WebTerminal"] = None - # 多智能体模式：为每个启用 multi_agent_mode 的会话维护一个 MultiAgentState + # 多智能体模式：MultiAgentState 是会话级运行态，全进程共享一份注册表 + # （见 modules/multi_agent/state.py 中 GLOBAL_MULTI_AGENT_STATES 的注释）， + # 避免多 manager 并存时同一对话被 from_snapshot 复制出多份独立副本。 # key = conversation_id, value = MultiAgentState - self.multi_agent_states: Dict[str, Any] = {} + self.multi_agent_states: Dict[str, Any] = GLOBAL_MULTI_AGENT_STATES self.base_dir.mkdir(parents=True, exist_ok=True) self.state_file.parent.mkdir(parents=True, exist_ok=True) @@ -1026,24 +1032,25 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi def get_or_create_multi_agent_state(self, conversation_id: str): """获取或为该会话创建 MultiAgentState。""" from modules.multi_agent.state import MultiAgentState - state = self.multi_agent_states.get(conversation_id) - if state: + with GLOBAL_MULTI_AGENT_STATES_LOCK: + state = self.multi_agent_states.get(conversation_id) + if state: + ma_debug( + "manager_get_or_create_ma_state_reuse", + conversation_id=conversation_id, + state_id=id(state), + manager_id=id(self), + ) + return state + state = MultiAgentState(conversation_id=conversation_id) + self.multi_agent_states[conversation_id] = state ma_debug( - "manager_get_or_create_ma_state_reuse", + "manager_get_or_create_ma_state_create", conversation_id=conversation_id, state_id=id(state), manager_id=id(self), ) return state - state = MultiAgentState(conversation_id=conversation_id) - self.multi_agent_states[conversation_id] = state - ma_debug( - "manager_get_or_create_ma_state_create", - conversation_id=conversation_id, - state_id=id(state), - manager_id=id(self), - ) - return state def get_multi_agent_state(self, conversation_id: str): """获取该会话的多智能体状态。""" @@ -1059,7 +1066,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi def drop_multi_agent_state(self, conversation_id: str) -> None: """删除会话状态（会话结束时调用）。""" - self.multi_agent_states.pop(conversation_id, None) + with GLOBAL_MULTI_AGENT_STATES_LOCK: + self.multi_agent_states.pop(conversation_id, None) def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int: """修正运行态任务状态。 diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index 0f52a6e7..3fba47ea 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -58,33 +58,97 @@ class SubAgentStateMixin: # 恢复多智能体运行态（如果状态文件包含） try: - from modules.multi_agent.state import MultiAgentState + from modules.multi_agent.state import ( + MultiAgentState, + GLOBAL_MULTI_AGENT_STATES_LOCK, + ) manager = self multi_agent_states = getattr(manager, "multi_agent_states", None) if multi_agent_states is not None and isinstance(multi_agent_states, dict): loaded_ma_states = data.get("multi_agent_states", {}) - for conv_id, snapshot in loaded_ma_states.items(): - try: - if isinstance(snapshot, dict): - # 关键：不要覆盖内存中已存在的 MultiAgentState， - # 否则 SubAgentTask 持有的旧引用上的 pending_master_messages - # 会被新的空 state 覆盖，导致子智能体输出丢失。 - if conv_id in multi_agent_states: + # multi_agent_states 是全进程共享注册表；加锁防止多个 manager + # 并发 _load_state 时对同一 conv 重复 from_snapshot 出多份副本。 + with GLOBAL_MULTI_AGENT_STATES_LOCK: + for conv_id, snapshot in loaded_ma_states.items(): + try: + if isinstance(snapshot, dict): + # 关键：不要覆盖内存中已存在的 MultiAgentState， + # 否则 SubAgentTask 持有的旧引用上的 pending_master_messages + # 会被新的空 state 覆盖，导致子智能体输出丢失。 + if conv_id in multi_agent_states: + ma_debug( + "load_state_skip_existing_ma_state", + conversation_id=conv_id, + existing_state_id=id(multi_agent_states[conv_id]), + ) + continue + multi_agent_states[conv_id] = MultiAgentState.from_snapshot(snapshot) ma_debug( - "load_state_skip_existing_ma_state", + "load_state_restore_ma_state", conversation_id=conv_id, - existing_state_id=id(multi_agent_states[conv_id]), + state_id=id(multi_agent_states[conv_id]), ) - continue - multi_agent_states[conv_id] = MultiAgentState.from_snapshot(snapshot) + except Exception as exc: + logger.warning(f"恢复多智能体状态失败 {conv_id}: {exc}") + # 用任务记录（持久真相）校准实例终态：terminate 只改了发起 manager + # 的内存与磁盘任务记录，而磁盘上的 ma 快照可能是更早的 idle 副本 + # （多 manager 并存时代其他 manager 会用陈旧内存快照覆盖落盘）。 + # 任务记录的 terminated/终态是吸收态，恢复快照后必须以此校准， + # 否则已终结实例会以 idle 复活显示。 + for cal_state in list(multi_agent_states.values()): + for agent in cal_state.list_all(): + cal_task = self.tasks.get(agent.task_id) + if not isinstance(cal_task, dict): + continue + cal_status = cal_task.get("status") + if ( + cal_status in TERMINAL_STATUSES.union({"terminated"}) + and agent.status != cal_status + ): ma_debug( - "load_state_restore_ma_state", - conversation_id=conv_id, - state_id=id(multi_agent_states[conv_id]), + "load_state_calibrate_agent_status", + conversation_id=getattr(cal_state, "conversation_id", ""), + agent_id=agent.agent_id, + before=agent.status, + after=cal_status, ) - except Exception as exc: - logger.warning(f"恢复多智能体状态失败 {conv_id}: {exc}") + cal_state.mark_status(agent.agent_id, cal_status) + # 存量脏数据自愈：历史版本曾把内部 agent_id 与角色内编号混用， + # 在编号缺失时生成过 "_None" 后缀显示名（如 Full-Stack Engineer_None）。 + # 这里按「对话 × 角色 × 任务创建时间」重新排序编号修复，幂等： + # 修完后 display_name 不再以 _None 结尾，不会再命中。 + none_agents = [] + for st in list(multi_agent_states.values()): + for agent in st.list_all(): + if str(agent.display_name or "").endswith("_None"): + none_agents.append((st, agent)) + if none_agents: + groups = {} + for st, agent in none_agents: + groups.setdefault((id(st), agent.role_id), []).append((st, agent)) + for (_sid, _role_id), items in groups.items(): + def _created_of(pair): + t = self.tasks.get(pair[1].task_id) or {} + return t.get("created_at") or 0 + items.sort(key=_created_of) + seq = 1 + for st, agent in items: + base = str(agent.display_name)[: -len("_None")] + if not base: + continue + new_name = f"{base}_{seq}" + seq += 1 + ma_debug( + "load_state_fix_none_display_name", + agent_id=agent.agent_id, + before=agent.display_name, + after=new_name, + ) + agent.display_name = new_name + task = self.tasks.get(agent.task_id) + if isinstance(task, dict) and str(task.get("display_name") or "").endswith("_None"): + task["display_name"] = new_name except Exception as exc: logger.warning(f"加载多智能体状态失败: {exc}") @@ -116,7 +180,8 @@ class SubAgentStateMixin: if multi_agent_states: payload["multi_agent_states"] = { conv_id: state.to_snapshot() - for conv_id, state in multi_agent_states.items() + # 全局共享注册表可能被其他线程并发增删，先拷贝再遍历 + for conv_id, state in list(multi_agent_states.items()) if isinstance(state, object) and hasattr(state, "to_snapshot") } except Exception as exc: