""" core/update_manager.py - 自动更新与热更新管理器 ================================================ 负责: 1. 定期检查远程版本（GitHub releases / npm registry / git tags） 2. 代码热更新: git pull → importlib.reload → 组件重建 3. 依赖热更新: pip install --upgrade（无进程重启） 4. 全量更新: git pull + pip install + 优雅重启（os.execv） 5. 利用现有 ConfigBroadcaster 协调 Agent 安全暂停/恢复 6. 利用 TaskQueue.drain() 确保运行中任务不受影响 安全保证: - 更新前自动暂停所有 Agent 任务（通过 ConfigBroadcaster） - 更新前排空任务队列（通过 TaskQueue.drain()） - 更新失败自动回滚 - 全量更新使用 os.execv 保持进程 PID 不变（systemd 友好） - 更新日志持久化，支持诊断 """ from __future__ import annotations import asyncio import importlib import json import os import subprocess import sys import time from dataclasses import dataclass, field from datetime import datetime from enum import Enum from pathlib import Path from typing import Any, Callable, Dict, List, Optional from core.logger import get_logger from core.version import get_version logger = get_logger("myagent.update") # 项目根目录 PROJECT_ROOT = Path(__file__).parent.parent class UpdateType(str, Enum): """更新类型""" CONFIG = "config" # 仅配置热重载（已有功能） CODE = "code" # 代码热更新（importlib.reload 模块） DEPENDENCY = "dependency" # 依赖更新（pip install --upgrade） FULL = "full" # 全量更新（git pull + 依赖 + 进程重启） class UpdateStatus(str, Enum): """更新状态""" IDLE = "idle" CHECKING = "checking" AVAILABLE = "available" DOWNLOADING = "downloading" UPDATING = "updating" DRAINING = "draining" # 排空任务队列 RELOADING = "reloading" # 热重载模块 RESTARTING = "restarting" # 进程重启 SUCCESS = "success" FAILED = "failed" ROLLBACK = "rollback" @dataclass class UpdateInfo: """远程版本信息""" current_version: str = "" latest_version: str = "" update_type: str = "" changelog: str = "" published_at: str = "" download_url: str = "" has_update: bool = False @dataclass class UpdateRecord: """更新记录（持久化到文件）""" id: str = "" type: str = "" status: str = "" from_version: str = "" to_version: str = "" started_at: str = "" finished_at: str = "" error: str = "" details: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict: return { "id": self.id, "type": self.type, "status": self.status, "from_version": self.from_version, "to_version": self.to_version, "started_at": self.started_at, "finished_at": self.finished_at, "error": self.error, "details": self.details, } class UpdateManager: """ 自动更新管理器。 核心能力: - 定时版本检查（GitHub / npm / PyPI） - 代码热更新（无需重启进程） - 全量更新（git pull + 优雅重启） - 更新安全门控（配合 ConfigBroadcaster + TaskQueue） - 更新历史记录 使用方式: # 初始化 update_mgr = UpdateManager( config_broadcaster=broadcaster, task_queue=task_queue, data_dir="~/.myagent", app_ref=app, # MyAgentApp 实例引用 ) await update_mgr.initialize() # 检查更新 info = await update_mgr.check_for_update() if info.has_update: await update_mgr.apply_update(UpdateType.FULL) # 定时检查（后台） await update_mgr.start_auto_check(interval=3600) 更新安全流程: 1. check_for_update() → 发现新版本 2. apply_update(type) → a. request_reload() via ConfigBroadcaster → 所有 Agent 暂停 b. task_queue.drain() → 等待运行中任务完成 c. 执行更新操作 d. complete_reload() → 所有 Agent 恢复 e. (FULL 类型) os.execv() → 优雅重启 """ def __init__( self, config_broadcaster=None, task_queue=None, data_dir: str = "", app_ref=None, ): # 外部依赖 self.config_broadcaster = config_broadcaster self.task_queue = task_queue self.app_ref = app_ref # MyAgentApp 实例引用 # ── [v1.26.8] 防循环更新机制 ── self._update_history = [] # 记录最近的更新历史 self._max_update_attempts = 3 # 最大尝试次数 self._last_update_check = {} # 版本检查去重 # 数据目录 self._data_dir = Path(data_dir) if data_dir else PROJECT_ROOT / ".update" self._data_dir.mkdir(parents=True, exist_ok=True) self._record_file = self._data_dir / "update_history.json" # 状态 self._status = UpdateStatus.IDLE self._current_update: Optional[UpdateRecord] = None self._last_check_time: float = 0 self._auto_check_task: Optional[asyncio.Task] = None self._auto_check_interval: int = 3600 # 默认 1 小时 self._auto_check_enabled: bool = False # 需要热重载的模块列表 self._reloadable_modules = [ "core.llm", "core.task_queue", "core.config_broadcast", "core.utils", "memory.manager", "executor.engine", "aiskills.registry", "agents.base", "agents.main_agent", "agents.tool_agent", "agents.memory_agent", ] # ── 属性 ── @property def status(self) -> str: return self._status.value @property def is_updating(self) -> bool: return self._status in ( UpdateStatus.DOWNLOADING, UpdateStatus.UPDATING, UpdateStatus.DRAINING, UpdateStatus.RELOADING, UpdateStatus.RESTARTING, ) # ── 初始化 ── async def initialize(self): """初始化更新管理器""" self._load_history() current_ver = get_version() logger.info(f"更新管理器已初始化 (当前版本: {current_ver})") # ── [v1.26.8] 检测更新后重启，延迟自动检查 ── # 如果是更新后重启，延迟检查更新，避免立即触发循环 if os.environ.get("MYAGENT_UPDATED") == "true": from_ver = os.environ.get("MYAGENT_UPDATED_FROM", "?") to_ver = os.environ.get("MYAGENT_UPDATED_TO", "?") logger.info(f"🔄 检测到更新后重启: v{from_ver} → v{to_ver}，延迟自动更新检查") # 清理环境变量 os.environ.pop("MYAGENT_UPDATED", None) os.environ.pop("MYAGENT_UPDATED_FROM", None) os.environ.pop("MYAGENT_UPDATED_TO", None) # 延迟检查（给用户时间观察更新结果，同时避免与启动冲突） asyncio.create_task(self._delayed_check_after_update()) async def _delayed_check_after_update(self): """更新后延迟检查更新（避免立即触发循环）""" try: # 等待 30 秒，让系统稳定 await asyncio.sleep(30) logger.info("更新后延迟检查开始...") info = await self.check_for_update() if info.has_update: logger.info(f"检测到新版本: {info.current_version} → {info.latest_version}，可通过管理后台手动更新") else: logger.info("当前已是最新版本") except Exception as e: logger.error(f"延迟更新检查失败: {e}") # ── 版本检查 ── async def check_for_update(self) -> UpdateInfo: """ 检查是否有新版本可用。 检查来源（按优先级）: 1. GitHub releases / tags 2. npm registry（如果是 npm 安装） 3. PyPI（如果是 pip 安装） 4. 本地 git（git fetch + tags） Returns: UpdateInfo: 版本比较结果 """ self._status = UpdateStatus.CHECKING self._last_check_time = time.time() current = get_version() try: # 尝试多种方式检查更新（npm 安装优先用 npm registry） if (PROJECT_ROOT / "package.json").exists(): latest = await self._check_npm_registry() if not latest: latest = await self._check_github_tags() if not latest: latest = await self._check_pypi() if not latest: latest = await self._check_local_git() if latest: has_update = self._version_compare(latest, current) > 0 else: has_update = False latest = current # ── [v1.26.8] 防循环更新检查 ── # 检查是否在短时间内重复尝试更新同一版本 check_key = f"{current}->{latest}" now = time.time() if check_key in self._last_update_check: last_time = self._last_update_check[check_key] if now - last_time < 300: # 5分钟内重复检测到同一更新 self._update_history.append({ 'timestamp': now, 'from': current, 'to': latest, 'attempt': self._update_history.count(check_key) + 1 }) # 如果尝试次数超过限制，暂时禁用更新提醒 attempts = sum(1 for h in self._update_history if h.get('attempt', 0) >= 3) if attempts >= self._max_update_attempts: logger.warning(f"⚠️ 检测到可能的更新循环（{attempts}次尝试），暂时禁用自动更新检查") # 延长检查间隔或跳过本次更新 has_update = False self._status = UpdateStatus.IDLE return UpdateInfo( current_version=current, latest_version=latest, has_update=False, ) # 记录本次检查 self._last_update_check[check_key] = now # 清理旧的检查记录（保留最近10条） if len(self._update_history) > 10: self._update_history = self._update_history[-10:] info = UpdateInfo( current_version=current, latest_version=latest, has_update=has_update, update_type="code" if has_update else "", ) self._status = UpdateStatus.AVAILABLE if has_update else UpdateStatus.IDLE logger.info( f"版本检查完成: 当前={current}, 最新={latest}, " f"有更新={has_update}" ) return info except Exception as e: self._status = UpdateStatus.IDLE logger.error(f"版本检查失败: {e}") return UpdateInfo( current_version=current, latest_version=current, has_update=False, ) async def _check_github_tags(self) -> str: """从 GitHub tags 检查最新版本""" try: # 检查是否在 git 仓库中 result = subprocess.run( ["git", "remote", "get-url", "origin"], capture_output=True, text=True, timeout=5, cwd=PROJECT_ROOT, ) remote_url = result.stdout.strip() if not remote_url: return "" # git fetch --tags subprocess.run( ["git", "fetch", "--tags", "--quiet"], capture_output=True, timeout=15, cwd=PROJECT_ROOT, ) # 获取最新 tag result = subprocess.run( ["git", "tag", "--sort=-v:refname"], capture_output=True, text=True, timeout=5, cwd=PROJECT_ROOT, ) tags = [t.strip() for t in result.stdout.strip().split("\n") if t.strip()] if not tags: return "" # 过滤出语义版本号 import re semver_tags = [] for tag in tags: m = re.match(r"^v?(\d+\.\d+\.\d+)", tag) if m: semver_tags.append((m.group(1), tag)) if semver_tags: # 返回最新的语义版本号 semver_tags.sort(key=lambda x: [int(p) for p in x[0].split(".")], reverse=True) return semver_tags[0][0] except (subprocess.TimeoutExpired, FileNotFoundError): pass except Exception as e: logger.debug(f"GitHub tag 检查失败: {e}") return "" async def _check_npm_registry(self) -> str: """从 npm registry 检查最新版本""" try: # 检查是否有 package.json pkg_json = PROJECT_ROOT / "package.json" if not pkg_json.exists(): return "" pkg_data = json.loads(pkg_json.read_text(encoding="utf-8")) pkg_name = pkg_data.get("name", "") if not pkg_name: return "" except Exception as e: logger.debug(f"读取 package.json 失败: {e}") return "" registry = self._get_npm_registry() # [v1.15.52] 优先检查官方源，镜像源可能有同步延迟 registries = ["https://registry.npmjs.org"] if registry != "https://registry.npmjs.org": registries.append(registry) # 直接请求 npm registry HTTP API（不依赖 npm 命令行） for reg in registries: try: import urllib.request url = f"{reg}/{pkg_name}/latest" logger.debug(f"检查 npm registry: {url}") loop = asyncio.get_running_loop() data = await loop.run_in_executor( None, lambda _url=url: urllib.request.urlopen(_url, timeout=15).read(), ) info = json.loads(data) version = info.get("version", "") if version and version[0].isdigit(): logger.debug(f"npm registry ({'official' if reg == registries[0] else 'mirror'}) 最新版本: {version}") return version except Exception as e: logger.debug(f"npm registry HTTP 检查失败 ({reg}): {e}") # 回退: 尝试用 npm 命令行检查 (可能 urllib 失败但 npm 命令可用) try: loop = asyncio.get_running_loop() npm_view_cmd = ["npm", "view", pkg_name, "version", "--registry", "https://registry.npmjs.org"] result = await loop.run_in_executor( None, lambda: subprocess.run( npm_view_cmd, capture_output=True, text=True, timeout=15, ) ) ver = result.stdout.strip() if ver and ver[0].isdigit(): logger.debug(f"npm view 最新版本: {ver}") return ver except Exception as e2: logger.debug(f"npm view 回退检查也失败: {e2}") return "" async def _check_pypi(self) -> str: """从 PyPI 检查最新版本""" try: loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, lambda: subprocess.run( ["pip", "index", "versions", "myagent"], capture_output=True, text=True, timeout=10, ) ) output = result.stdout.strip() if output and "available" in output.lower(): import re m = re.search(r"(\d+\.\d+\.\d+)", output) if m: return m.group(1) except (subprocess.TimeoutExpired, FileNotFoundError): pass except Exception as e: logger.debug(f"PyPI 检查失败: {e}") return "" async def _check_local_git(self) -> str: """检查本地 git 最新 tag（离线模式）""" try: result = subprocess.run( ["git", "describe", "--tags", "--abbrev=0"], capture_output=True, text=True, timeout=5, cwd=PROJECT_ROOT, ) tag = result.stdout.strip().lstrip("v") import re m = re.match(r"(\d+\.\d+\.\d+)", tag) if m: return m.group(1) except Exception: pass return "" # ── 应用更新 ── async def apply_update( self, update_type: UpdateType = UpdateType.FULL, target_version: str = "", ) -> UpdateRecord: """ 执行更新。 安全流程: 1. 通过 ConfigBroadcaster 通知所有 Agent 暂停 2. 通过 TaskQueue.drain() 等待运行中任务完成 3. 执行更新操作 4. 重建/刷新组件 5. 通知 Agent 恢复 6. (FULL) os.execv 优雅重启 Args: update_type: 更新类型 target_version: 目标版本（空=最新） Returns: UpdateRecord: 更新记录 """ if self.is_updating: raise RuntimeError(f"更新正在进行中 (状态: {self._status.value})") current = get_version() record = UpdateRecord( id=f"upd_{int(time.time())}", type=update_type.value, from_version=current, to_version=target_version or "latest", started_at=datetime.now().isoformat(), status="draining", ) self._current_update = record try: # ── Phase 1: 安全门控 - 暂停 Agent + 排空队列 ── await self._enter_safe_mode(update_type) # ── Phase 2: 执行更新 ── if update_type == UpdateType.CODE: await self._do_code_update(record) elif update_type == UpdateType.DEPENDENCY: await self._do_dependency_update(record) elif update_type == UpdateType.FULL: await self._do_full_update(record) elif update_type == UpdateType.CONFIG: await self._do_config_update(record) else: raise ValueError(f"不支持的更新类型: {update_type}") record.status = "success" record.finished_at = datetime.now().isoformat() except Exception as e: logger.error(f"更新失败: {e}", exc_info=True) record.status = "failed" record.error = str(e) record.finished_at = datetime.now().isoformat() # 尝试回滚 try: await self._rollback(record) except Exception as rb_err: logger.error(f"回滚也失败了: {rb_err}") finally: self._status = UpdateStatus.SUCCESS if record.status == "success" else UpdateStatus.FAILED # 保存记录 self._save_record(record) self._current_update = None # 恢复 Agent # 注意: FULL 类型成功时 os.execv 会替换进程，finally 不会执行 # 只有失败时才会走到这里，所以必须恢复 Agent try: await self._exit_safe_mode() except Exception as exit_err: logger.error(f"退出安全模式失败: {exit_err}") return record async def _enter_safe_mode(self, update_type: UpdateType = UpdateType.CONFIG): """ 进入安全模式: 暂停所有 Agent 任务 + 排空任务队列。 利用现有的 ConfigBroadcaster 和 TaskQueue 机制。 Args: update_type: 更新类型，影响 Agent 等待超时时间 FULL 类型给 Agent 120s 等待，其他 30s """ self._status = UpdateStatus.DRAINING logger.info("进入更新安全模式...") # 映射 UpdateType → ReloadType，确保 Agent 等待正确的超时时间 from core.config_broadcast import ReloadType reload_type_map = { UpdateType.CONFIG: ReloadType.CONFIG, UpdateType.CODE: ReloadType.CODE, UpdateType.DEPENDENCY: ReloadType.DEPENDENCY, UpdateType.FULL: ReloadType.FULL, } reload_type = reload_type_map.get(update_type, ReloadType.CONFIG) # Step 1: 通过 ConfigBroadcaster 请求所有 Agent 暂停 if self.config_broadcaster: paused_count = await self.config_broadcaster.request_reload(reload_type=reload_type) logger.info(f"已通知 {paused_count} 个 Agent 暂停 (类型={reload_type.value})") # Step 2: 排空任务队列（等待运行中任务完成，拒绝新任务） if self.task_queue: try: await self.task_queue.drain(timeout=60) logger.info("任务队列已排空") except asyncio.TimeoutError: logger.warning("任务队列排空超时，部分任务可能仍在运行") logger.info("安全模式就绪，开始执行更新...") async def _exit_safe_mode(self): """退出安全模式: 通知所有 Agent 恢复""" if self.config_broadcaster: await self.config_broadcaster.complete_reload() self._status = UpdateStatus.IDLE logger.info("已退出安全模式") async def _do_code_update(self, record: UpdateRecord): """ 代码热更新: git pull + importlib.reload + 组件重建。 热更新策略: - 对纯数据模块（config, version）: importlib.reload 即可 - 对含状态的模块（LLM client, memory）: reload 后刷新实例属性 - 对 Agent 模块: reload 后通过 app_ref 重建 Agent 实例 """ self._status = UpdateStatus.UPDATING logger.info("开始代码热更新...") # Step 1: git pull 拉取最新代码 logger.info("执行 git pull...") result = subprocess.run( ["git", "pull", "--ff-only"], capture_output=True, text=True, timeout=30, cwd=PROJECT_ROOT, ) if result.returncode != 0: # 可能是 fast-forward 失败，尝试普通 pull result = subprocess.run( ["git", "pull"], capture_output=True, text=True, timeout=30, cwd=PROJECT_ROOT, ) pull_output = result.stdout + result.stderr logger.info(f"git pull 完成: {pull_output[:500]}") # Step 2: 热重载模块 self._status = UpdateStatus.RELOADING await self._reload_modules() # Step 3: 刷新组件引用 self._refresh_components() record.to_version = get_version() record.details["pull_output"] = pull_output[:1000] logger.info("代码热更新完成") async def _do_dependency_update(self, record: UpdateRecord): """依赖更新: pip install --upgrade""" self._status = UpdateStatus.UPDATING logger.info("开始依赖更新...") # 使用当前 Python 解释器 loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, lambda: subprocess.run( [sys.executable, "-m", "pip", "install", "--upgrade", "-r", str(PROJECT_ROOT / "requirements.txt")], capture_output=True, text=True, timeout=300, cwd=PROJECT_ROOT, ) ) pip_output = result.stdout + result.stderr record.details["pip_output"] = pip_output[:2000] if result.returncode != 0: raise RuntimeError(f"pip install 失败: {pip_output[-500:]}") logger.info("依赖更新完成") async def _do_full_update(self, record: UpdateRecord): """ 全量更新: npm install -g (npm 安装) 或 git pull + pip install (源码安装) + 优雅重启。 使用 os.execv 替换当前进程，保持 PID 不变。 """ self._status = UpdateStatus.UPDATING logger.info("开始全量更新...") is_npm_install = self._is_npm_install() if is_npm_install: # npm 安装: npm install -g myagent-ai await self._npm_global_update(record) else: # 源码安装: git pull + pip install await self._source_update(record) # 验证更新是否成功 — npm 安装后需从 npm 全局目录读取新版本 # 因为当前进程的 PROJECT_ROOT 指向旧版本，get_version() 会读到旧版本号 new_version = self._read_installed_version() or get_version() if new_version == record.from_version: # [v1.16.7] 修复：旧条件 `record.to_version != "latest"` 永远为 False # 因为 record.to_version 初始值就是 "latest"，导致版本未变化时不会报警 logger.warning( f"更新后版本仍为 {new_version}，更新可能未生效！" f"(npm路径={is_npm_install})" ) # 不直接抛异常，让重启继续尝试（重启后 start.js 会重新检测版本和依赖） # Step 3: 保存更新完成标记（用于重启后验证） record.status = "restarting" record.to_version = new_version self._save_record(record) logger.info(f"全量更新完成 v{record.from_version} → v{new_version}，准备优雅重启...") # Step 4: 清除旧的 .pyc 缓存（防止更新后仍加载旧字节码） self._clear_pycache() # Step 5: 优雅重启 self._status = UpdateStatus.RESTARTING await asyncio.sleep(0.5) # 给日志一点时间写入 # 传递特殊环境变量标识这是更新后重启 env = os.environ.copy() env["MYAGENT_UPDATED"] = "true" env["MYAGENT_UPDATED_FROM"] = record.from_version env["MYAGENT_UPDATED_TO"] = record.to_version if is_npm_install: # npm 安装: 通过 start.js (Node.js) 重启，确保 venv 和依赖正确处理 # [v1.15.54] 关键修复: 必须从 npm 全局安装目录找到新版本的 start.js # 因为 PROJECT_ROOT 指向旧版本目录，用旧 start.js 重启等于没更新 import shutil node_cmd = shutil.which("node") new_start_js = self._find_npm_start_js() if new_start_js: start_js = new_start_js elif (PROJECT_ROOT / "start.js").exists(): # [v1.16.7] 只有当 PROJECT_ROOT 下的 start.js 版本确实更新了才使用 # 如果版本未变，说明 npm install 没有更新 PROJECT_ROOT logger.warning( f"未在 npm 全局目录找到新版 start.js，" f"回退使用 PROJECT_ROOT: {PROJECT_ROOT / 'start.js'}" ) start_js = PROJECT_ROOT / "start.js" else: logger.error(f"未找到任何 start.js 文件！") start_js = None if start_js and start_js.exists(): if node_cmd: # 提取原始参数 (去掉 main.py，还原 start.js 的参数格式) original_args = [] skip_next = False for arg in sys.argv[1:]: if skip_next: skip_next = False continue if arg == "--web": original_args.append("web") elif arg == "--tray": original_args.append("tray") elif arg == "--cli": original_args.append("cli") elif arg == "--server": original_args.append("server") elif arg == "--setup": original_args.append("setup") elif arg.startswith("--port="): original_args.append(arg.replace("--port=", "")) elif arg.startswith("--"): skip_next = True # 跳过 --arg value 形式的参数 logger.info(f"通过 start.js 重启: node {start_js} {' '.join(original_args)}") os.execv(node_cmd, [node_cmd, str(start_js)] + original_args) else: logger.warning("未找到 node 命令，回退到直接 Python 重启") else: logger.error(f"未找到 start.js (尝试: {new_start_js}, 回退: {PROJECT_ROOT / 'start.js'})") # 回退: 直接用 Python 重启（源码安装或 node 不可用） # 使用绝对路径避免 cwd 变化导致找不到 main.py abs_main = str((PROJECT_ROOT / "main.py").resolve()) logger.info(f"通过 Python 直接重启: {sys.executable} {abs_main}") os.execv(sys.executable, [sys.executable, abs_main] + sys.argv[1:]) def _is_npm_install(self) -> bool: """ 判断是否通过 npm 全局安装。 不仅检查 package.json 是否存在（source 安装也有）， 还检查是否位于 npm 全局目录中。 """ pkg_json = PROJECT_ROOT / "package.json" if not pkg_json.exists(): return False # 检查是否在 npm 全局目录中 try: result = subprocess.run( ["npm", "root", "-g"], capture_output=True, text=True, timeout=5, ) npm_root = result.stdout.strip() if npm_root: # 规范化路径比较 import pathlib project_resolved = PROJECT_ROOT.resolve() npm_root_resolved = pathlib.Path(npm_root).resolve() # 检查 PROJECT_ROOT 是否在 npm 全局目录下 try: project_resolved.relative_to(npm_root_resolved) return True except ValueError: pass # 也检查包名子目录 (npm root -g / myagent-ai) pkg_dir = npm_root_resolved / "myagent-ai" if project_resolved == pkg_dir: return True except (subprocess.TimeoutExpired, FileNotFoundError): pass # npm 命令不可用时，检查是否有 .git 目录来区分 # 有 .git = 源码安装，没有 = 可能是 npm 安装 if (PROJECT_ROOT / ".git").exists(): return False # 默认假设 npm 安装（保守策略） logger.debug("无法确定安装方式，默认使用 npm 更新路径") return True def _read_installed_version(self) -> str: """[v1.15.52] 从 npm 全局安装目录读取实际安装的最新版本号。 npm install -g 会将新版本安装到 npm 全局目录（如 /root/.npm/.../node_modules/myagent-ai/）， 但当前进程的 PROJECT_ROOT 仍然指向旧版本的目录，get_version() 会读到旧版本。 此方法通过 npm root -g 找到全局安装目录，从其中的 package.json 读取真实版本。 """ try: result = subprocess.run( ["npm", "root", "-g"], capture_output=True, text=True, timeout=5, ) npm_root = result.stdout.strip() if not npm_root: return "" pkg_json = Path(npm_root) / "myagent-ai" / "package.json" if pkg_json.exists(): data = json.loads(pkg_json.read_text(encoding="utf-8")) ver = data.get("version", "") if ver: logger.debug(f"从 npm 全局目录读取到版本: {ver} (路径: {pkg_json})") return ver except Exception as e: logger.debug(f"读取 npm 全局安装版本失败: {e}") return "" def _find_npm_start_js(self) -> Optional[Path]: """[v1.15.54] 从 npm 全局安装目录找到新版本的 start.js 路径。 npm install -g 安装的新版本在 npm root -g/myagent-ai/ 下， 但当前进程的 PROJECT_ROOT 可能指向旧的 npx 缓存目录。 重启时必须用新版本的 start.js，否则等于没更新。 [v1.16.7] 增加多种查找方式：npm root -g / npm bin -g / which， 确保找到新版本的 start.js，避免 fallback 到旧代码。 """ candidates = [] # 方法1: npm root -g / myagent-ai / start.js try: result = subprocess.run( ["npm", "root", "-g"], capture_output=True, text=True, timeout=5, ) npm_root = result.stdout.strip() if npm_root: candidates.append(Path(npm_root) / "myagent-ai" / "start.js") except Exception: pass # 方法2: npm bin -g / myagent-ai (symlink) try: result = subprocess.run( ["npm", "bin", "-g"], capture_output=True, text=True, timeout=5, ) npm_bin = result.stdout.strip() if npm_bin: candidates.append(Path(npm_bin) / "myagent-ai") except Exception: pass # 方法3: which myagent-ai (全局命令路径，通常是 symlink) try: result = subprocess.run( ["which", "myagent-ai"], capture_output=True, text=True, timeout=5, ) which_path = result.stdout.strip() if which_path: # which 返回的可能是 symlink，需要 resolve candidates.append(Path(which_path).resolve()) except Exception: pass # 方法4: 直接读取 npm list -g 输出 try: result = subprocess.run( ["npm", "list", "-g", "myagent-ai", "--json"], capture_output=True, text=True, timeout=10, ) if result.returncode == 0: import json as _json data = _json.loads(result.stdout) resolved = data.get("dependencies", {}).get("myagent-ai", {}).get("resolved", "") if resolved: candidates.append(Path(resolved) / "start.js") except Exception: pass # 按优先级尝试 for candidate in candidates: try: if candidate.exists(): # 读取该 start.js 对应的 package.json 版本 pkg_dir = candidate.parent pkg_json = pkg_dir / "package.json" if pkg_json.exists(): data = json.loads(pkg_json.read_text(encoding="utf-8")) ver = data.get("version", "") logger.info(f"找到 npm start.js: {candidate} (v{ver})") return candidate else: logger.info(f"找到 npm start.js: {candidate} (无 package.json)") return candidate except Exception: continue logger.warning( f"未找到 npm 全局安装的新版 start.js " f"(已尝试 {len(candidates)} 个候选路径)" ) return None def _get_npm_registry(self) -> str: """获取 npm registry 地址，国内用户自动使用镜像""" import locale lang = (os.environ.get("LANG", "") or os.environ.get("LC_ALL", "") or "").lower() env_mirror = os.environ.get("MYAGENT_NPM_MIRROR", "") if env_mirror in ("1", "true", "cn"): return "https://registry.npmmirror.com" if lang and ("zh_cn" in lang or "chinese" in lang): return "https://registry.npmmirror.com" # 通过 IP 地址判断（可选） return "https://registry.npmjs.org" async def _npm_global_update(self, record: UpdateRecord): """执行 npm 全局包更新""" self._status = UpdateStatus.DOWNLOADING registry = self._get_npm_registry() loop = asyncio.get_running_loop() pkg_data = json.loads((PROJECT_ROOT / "package.json").read_text(encoding="utf-8")) pkg_name = pkg_data.get("name", "myagent-ai") # [v1.15.52] 先尝试官方源安装，确保拉到最新版本 # 镜像源可能有同步延迟导致安装旧版本 registries_to_try = ["https://registry.npmjs.org"] if registry != "https://registry.npmjs.org": registries_to_try.append(registry) # 镜像作为备用 # E404 重试: 新版本刚发布时 CDN 可能尚未传播，最多等待 3 分钟 max_404_retries = 6 retry_interval = 30 # 秒 last_error = None for reg in registries_to_try: is_official = (reg == "https://registry.npmjs.org") # 构建 npm install 命令 — 使用 @latest + --prefer-online 确保拉取最新版本（不走缓存） npm_cmd = ["npm", "install", "-g", f"{pkg_name}@latest", "--prefer-online", "--registry", reg] if is_official: logger.info(f"使用 npm 官方源安装最新版本: {reg}") else: logger.info(f"使用 npm 镜像源安装: {reg}") result = await loop.run_in_executor( None, lambda cmd=npm_cmd: subprocess.run( cmd, capture_output=True, text=True, timeout=300, ) ) npm_output = result.stdout + result.stderr record.details["update_output"] = npm_output[:2000] logger.info(f"npm update ({'official' if is_official else 'mirror'}): {npm_output[:500]}") if result.returncode != 0: error_msg = npm_output[-500:] if npm_output else "未知错误" # 检查是否包含真正的错误 has_real_error = False is_404 = False for line in error_msg.split("\n"): stripped = line.strip() if stripped.startswith("ERR!") or "code E" in stripped: has_real_error = True if "code E404" in stripped or "Not Found" in stripped and pkg_name in stripped: is_404 = True if has_real_error: last_error = error_msg if is_404: # E404: 新版本尚未传播到该 registry，等待后重试 remaining = max_404_retries while remaining > 0: remaining -= 1 logger.warning( f"npm install 通过 {'官方源' if is_official else '镜像源'} 失败 (E404，新版本尚未传播)，" f"{retry_interval}秒后重试 (剩余 {remaining} 次)..." ) await asyncio.sleep(retry_interval) result = await loop.run_in_executor( None, lambda cmd=npm_cmd: subprocess.run( cmd, capture_output=True, text=True, timeout=300, ) ) npm_output = result.stdout + result.stderr record.details["update_output"] = npm_output[:2000] logger.info(f"npm update retry ({'official' if is_official else 'mirror'}): {npm_output[:500]}") if result.returncode == 0 or "code E404" not in npm_output: # 成功或非 404 错误，跳出重试 break else: # 重试耗尽，放弃该源 logger.error(f"npm install E404 重试 {max_404_retries} 次仍失败，跳过该源") continue else: logger.warning(f"npm install 通过 {'官方源' if is_official else '镜像源'} 失败，尝试{'镜像源' if is_official else '官方源'}") continue else: logger.warning(f"npm install 有警告但可能已成功: {error_msg}") # 安装成功或只有警告，跳出循环 break if last_error and not (result.returncode == 0 or not any( line.strip().startswith("ERR!") for line in (npm_output.split("\n")[-20:]) )): raise RuntimeError(f"npm install -g {pkg_name} 所有源均失败:\n{last_error}") async def _source_update(self, record: UpdateRecord): """执行源码安装更新 (git pull + pip install) [v1.16.7] 当 git pull 因本地修改失败时，自动 stash → pull → pop， 避免因未提交的修改导致更新永远失败。 """ # git pull result = subprocess.run( ["git", "pull"], capture_output=True, text=True, timeout=30, cwd=PROJECT_ROOT, ) pull_output = result.stdout + result.stderr logger.info(f"git pull: {pull_output[:500]}") # [v1.16.7] git pull 失败时，尝试 stash 本地修改后重新 pull if result.returncode != 0: logger.warning("git pull 失败，尝试 stash 本地修改后重新拉取...") # Stash 所有本地修改（包括未跟踪文件） stash_result = subprocess.run( ["git", "stash", "push", "-u", "-m", "auto-update-stash"], capture_output=True, text=True, timeout=10, cwd=PROJECT_ROOT, ) if stash_result.returncode == 0: logger.info(f"已 stash 本地修改: {stash_result.stdout.strip()}") # 重新尝试 git pull result = subprocess.run( ["git", "pull"], capture_output=True, text=True, timeout=30, cwd=PROJECT_ROOT, ) pull_output = result.stdout + result.stderr logger.info(f"git pull (stash后): {pull_output[:500]}") if result.returncode == 0: # pull 成功，尝试恢复 stash（可能冲突，静默丢弃） pop_result = subprocess.run( ["git", "stash", "pop"], capture_output=True, text=True, timeout=10, cwd=PROJECT_ROOT, ) if pop_result.returncode != 0: logger.warning( f"恢复 stash 失败（本地修改与新代码冲突），已丢弃: " f"{pop_result.stderr[:200]}" ) subprocess.run( ["git", "stash", "drop"], capture_output=True, text=True, timeout=5, cwd=PROJECT_ROOT, ) else: # pull 仍然失败，恢复 stash logger.error(f"stash 后 git pull 仍然失败: {pull_output[:300]}") subprocess.run( ["git", "stash", "pop"], capture_output=True, text=True, timeout=10, cwd=PROJECT_ROOT, ) raise RuntimeError( f"git pull 失败（即使 stash 后）: {pull_output[-300:]}" ) else: raise RuntimeError( f"git pull 失败且无法 stash: " f"pull={pull_output[-200:]}, stash={stash_result.stderr[-200:]}" ) # pip install self._status = UpdateStatus.DOWNLOADING loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, lambda: subprocess.run( [sys.executable, "-m", "pip", "install", "-e", ".", "--quiet"], capture_output=True, text=True, timeout=300, cwd=PROJECT_ROOT, ) ) pip_output = result.stdout + result.stderr record.details["update_output"] = pip_output[:1000] if result.returncode != 0: raise RuntimeError(f"pip install 失败: {pip_output[-300:]}") # ── [v1.26.8] 修复更新循环：重新加载版本模块 ── # pip install 后，新代码已安装，但 Python 模块仍缓存旧版本 # 必须重新加载 core.version 才能获取新版本号，避免重启后再次检查时陷入循环 try: import core.version importlib.reload(core.version) # 更新本地的版本缓存（如果存在） if 'core.version' in sys.modules: new_ver = core.version.get_version() logger.info(f"✅ 版本模块已重载，新版本: {new_ver}") # 如果 UpdateManager 有缓存版本，也更新它 if hasattr(self, '_current_version'): self._current_version = new_ver except Exception as e: logger.warning(f"版本模块重载失败（将使用重启后版本）: {e}") async def _do_config_update(self, record: UpdateRecord): """仅配置热重载（利用现有 ConfigManager.reload()）""" self._status = UpdateStatus.RELOADING logger.info("执行配置热重载...") if self.app_ref: # 使用现有的配置管理器重载 self.app_ref.config_mgr.reload() # 刷新 LLM 客户端 if self.app_ref.llm: new_cfg = self.app_ref.config.llm self.app_ref.llm.provider = new_cfg.provider self.app_ref.llm.model = new_cfg.model self.app_ref.llm.api_key = new_cfg.api_key self.app_ref.llm.base_url = new_cfg.base_url self.app_ref.llm._client = None # 强制重建 logger.info("配置热重载完成") else: # 没有 app 引用，直接 reload config 模块 if "config" in sys.modules: importlib.reload(sys.modules["config"]) logger.info("config 模块已重载") def _clear_pycache(self): """[v1.15.93] 清除项目目录和 npm 全局目录中的 __pycache__， 防止更新后 Python 仍加载旧的 .pyc 字节码文件。""" import shutil dirs_to_clean = [PROJECT_ROOT] # 也清除 npm 全局目录的缓存 try: result = subprocess.run( ["npm", "root", "-g"], capture_output=True, text=True, timeout=5, ) npm_root = result.stdout.strip() if npm_root: npm_pkg_dir = Path(npm_root) / "myagent-ai" if npm_pkg_dir.exists() and npm_pkg_dir not in dirs_to_clean: dirs_to_clean.append(npm_pkg_dir) except Exception: pass for base_dir in dirs_to_clean: for pycache_dir in base_dir.rglob("__pycache__"): try: shutil.rmtree(pycache_dir, ignore_errors=True) logger.debug(f"已清除缓存: {pycache_dir}") except Exception: pass logger.info(f"__pycache__ 清理完成 (检查了 {len(dirs_to_clean)} 个目录)") # ── 模块热重载 ── async def _reload_modules(self): """ 热重载 Python 模块。 按依赖顺序重载，确保被依赖的模块先重载。 """ reload_errors = [] # 重载顺序: 基础 → 复杂 reload_order = [ # 基础模块 "core.version", "core.utils", "core.logger", # 核心模块 "core.config_broadcast", "core.task_queue", "core.llm", # 功能模块 "memory.manager", "executor.engine", "aiskills.registry", # Agent 模块 "agents.base", "agents.memory_agent", "agents.tool_agent", "agents.main_agent", ] for module_name in reload_order: if module_name in sys.modules: try: module = sys.modules[module_name] importlib.reload(module) logger.debug(f"模块已重载: {module_name}") except Exception as e: error_msg = f"模块 {module_name} 重载失败: {e}" logger.warning(error_msg) reload_errors.append(error_msg) if reload_errors: logger.warning(f"部分模块重载失败 ({len(reload_errors)} 个)") def _refresh_components(self): """ 刷新组件实例属性，使重载后的代码生效。 重建 Agent 实例是最安全的做法。 """ if not self.app_ref: logger.warning("无 app 引用，跳过组件刷新") return app = self.app_ref try: # 刷新 LLM 客户端（强制重建底层 SDK 连接） if app.llm: app.llm._client = None logger.info("LLM 客户端已标记重建") # 重建技能注册表 if app.skill_registry: from aiskills.registry import SkillRegistry old_skills = app.skill_registry app.skill_registry = SkillRegistry() app._register_builtin_skills() logger.info(f"技能注册表已重建 ({len(app.skill_registry.list_skills())} 个技能)") # 重建 Agent 实例（保留引用关系） if app.tool_agent: from agents.tool_agent import ToolAgent app.tool_agent = ToolAgent( llm=app.llm, memory_manager=app.memory, executor=app.executor, skill_registry=app.skill_registry, task_queue=app.task_queue, config=app.config, config_broadcaster=app.config_broadcaster, ) if app.memory_agent: from agents.memory_agent import MemoryAgent app.memory_agent = MemoryAgent( llm=app.llm, memory_manager=app.memory, config=app.config, config_broadcaster=app.config_broadcaster, ) if app.main_agent: from agents.main_agent import MainAgent app.main_agent = MainAgent( llm=app.llm, memory_manager=app.memory, executor=app.executor, skill_registry=app.skill_registry, task_queue=app.task_queue, config=app.config, tool_agent=app.tool_agent, memory_agent=app.memory_agent, config_broadcaster=app.config_broadcaster, ) logger.info("Agent 实例已重建") except Exception as e: logger.error(f"组件刷新失败: {e}") async def _rollback(self, record: UpdateRecord): """回滚: 重新拉取之前的版本""" logger.warning(f"开始回滚更新...") try: # git reset --hard 到上一个版本 subprocess.run( ["git", "reset", "--hard", "HEAD~1"], capture_output=True, text=True, timeout=15, cwd=PROJECT_ROOT, ) record.status = "rollback" record.finished_at = datetime.now().isoformat() logger.info("回滚完成") except Exception as e: logger.error(f"回滚失败: {e}") # ── 定时自动检查 ── async def start_auto_check(self, interval: int = 3600): """ 启动后台定时版本检查。 Args: interval: 检查间隔（秒），默认 1 小时 """ self._auto_check_interval = interval self._auto_check_enabled = True if self._auto_check_task and not self._auto_check_task.done(): self._auto_check_task.cancel() self._auto_check_task = asyncio.create_task(self._auto_check_loop()) logger.info(f"自动版本检查已启动 (间隔: {interval}s)") async def stop_auto_check(self): """停止后台定时检查""" self._auto_check_enabled = False if self._auto_check_task and not self._auto_check_task.done(): self._auto_check_task.cancel() try: await self._auto_check_task except asyncio.CancelledError: pass logger.info("自动版本检查已停止") async def _auto_check_loop(self): """后台自动检查循环""" # [v1.15.97] 启动后立即执行一次检查，不等间隔 try: logger.info("执行启动时版本检查...") info = await self.check_for_update() if info.has_update: logger.info( f"发现新版本: {info.latest_version} " f"(当前: {info.current_version})" ) try: await self.apply_update(UpdateType.FULL) except Exception as e: logger.error(f"自动更新失败: {e}") except asyncio.CancelledError: return except Exception as e: logger.error(f"启动时检查异常: {e}") while self._auto_check_enabled: try: await asyncio.sleep(self._auto_check_interval) if not self._auto_check_enabled: break logger.info("执行定时版本检查...") info = await self.check_for_update() if info.has_update: logger.info( f"发现新版本: {info.latest_version} " f"(当前: {info.current_version})" ) # 自动应用全量更新 try: await self.apply_update(UpdateType.FULL) except Exception as e: logger.error(f"自动更新失败: {e}") except asyncio.CancelledError: break except Exception as e: logger.error(f"自动检查异常: {e}") await asyncio.sleep(60) # 出错后等 1 分钟再试 # ── 更新历史记录 ── def _load_history(self): """加载更新历史""" if self._record_file.exists(): try: self._history = json.loads( self._record_file.read_text(encoding="utf-8") ) except Exception: self._history = [] else: self._history = [] def _save_record(self, record: UpdateRecord): """保存更新记录""" self._load_history() self._history.insert(0, record.to_dict()) # 只保留最近 50 条记录 self._history = self._history[:50] try: self._record_file.write_text( json.dumps(self._history, indent=2, ensure_ascii=False), encoding="utf-8", ) except Exception as e: logger.error(f"保存更新记录失败: {e}") def get_history(self, limit: int = 20) -> List[dict]: """获取更新历史""" self._load_history() return self._history[:limit] # ── 状态查询 ── def get_status(self) -> Dict[str, Any]: """获取更新管理器完整状态""" return { "status": self._status.value, "current_version": get_version(), "is_updating": self.is_updating, "auto_check_enabled": self._auto_check_enabled, "auto_check_interval": self._auto_check_interval, "last_check_time": self._last_check_time, "current_update": self._current_update.to_dict() if self._current_update else None, } # ── 工具方法 ── @staticmethod def _version_compare(v1: str, v2: str) -> int: """ 比较两个语义版本号。 Returns: >0 if v1 > v2, 0 if equal, <0 if v1 < v2 """ import re def parse(v: str) -> List[int]: m = re.match(r"(\d+)\.(\d+)\.(\d+)", v) if m: return [int(m.group(1)), int(m.group(2)), int(m.group(3))] return [0, 0, 0] p1, p2 = parse(v1), parse(v2) if p1 > p2: return 1 elif p1 < p2: return -1 return 0