fix(multi-agent): MultiAgentState 全局单例，根治 terminate 后状态分裂复活
- 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）
This commit is contained in:
parent
171833e4d1
commit
77a271fa23
@ -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` 中不得混用。
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -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", ""))
|
||||
|
||||
@ -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()
|
||||
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()
|
||||
@ -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:
|
||||
"""修正运行态任务状态。
|
||||
|
||||
@ -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:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user