From 5d34b247ef71383344ebf79eee2b65f14a6edf32 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Thu, 16 Jul 2026 18:55:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20xiaohe=20upgrade=20+=20=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E5=99=A8=E6=9B=B4=E5=90=8D=20Xiaohe=20Agent=20+=20CLI?= =?UTF-8?q?=20=E8=BE=93=E5=87=BA=E5=85=A8=E8=8B=B1=E6=96=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - upgrade: 查版本(install.sh)→确认→rich 进度条下载→落 runtime 切 current(留两版)→重装三件套→自动 sync - launcher: 'Xiaohe Agent.app' / 'Xiaohe Agent.desktop'，自动清理旧 XiaoheAgent.{command,app,desktop} - 修复 _current_version 对缺失 current 软链误判为 'current' - 全部提示/log 英文化；93 tests passed --- myagents/commands/sync_workspace.py | 32 ++-- myagents/commands/upgrade.py | 238 ++++++++++++++++++++++++++++ myagents/entrypoints.py | 4 +- tests/test_sync_workspace.py | 30 ++-- tests/test_upgrade.py | 134 ++++++++++++++++ 5 files changed, 409 insertions(+), 29 deletions(-) create mode 100644 myagents/commands/upgrade.py create mode 100644 tests/test_upgrade.py diff --git a/myagents/commands/sync_workspace.py b/myagents/commands/sync_workspace.py index 9a9ca64..cd11796 100644 --- a/myagents/commands/sync_workspace.py +++ b/myagents/commands/sync_workspace.py @@ -197,12 +197,15 @@ def _print_report(workspace: Path, report: dict) -> None: console.print(f" [dim]- {rel}[/dim]") +_LAUNCHER_NAME = "Xiaohe Agent" +_LEGACY_LAUNCHER_NAMES = ("XiaoheAgent.command", "XiaoheAgent.app", "XiaoheAgent.desktop") + _APP_INFO_PLIST = """ - CFBundleNameXiaoheAgent - CFBundleDisplayNameXiaoheAgent + CFBundleNameXiaohe Agent + CFBundleDisplayNameXiaohe Agent CFBundleIdentifiercom.xiaohe.agent CFBundleVersion1 CFBundlePackageTypeAPPL @@ -215,7 +218,7 @@ _APP_INFO_PLIST = """ """ _APP_EXECUTABLE = """#!/usr/bin/env bash -# XiaoheAgent launcher — opens Terminal running xiaohe. +# Xiaohe Agent launcher — opens Terminal running xiaohe. export PATH="$HOME/.local/bin:/opt/homebrew/bin:/usr/local/bin:$PATH" if command -v xiaohe >/dev/null 2>&1; then osascript -e 'tell application "Terminal" to activate' \\ @@ -237,6 +240,15 @@ def _icon_assets() -> tuple[Path | None, Path | None]: return (icns if icns.is_file() else None), (png if png.is_file() else None) +def _remove_legacy_launchers(directory: Path) -> None: + for name in _LEGACY_LAUNCHER_NAMES: + legacy = directory / name + if legacy.is_dir(): + shutil.rmtree(legacy) + elif legacy.exists(): + legacy.unlink() + + def _create_launcher() -> None: """Double-click launcher that opens a terminal running ``xiaohe``.""" home = Path.home() @@ -245,11 +257,8 @@ def _create_launcher() -> None: desktop = home / "Desktop" if not desktop.is_dir(): return - # Legacy .command launcher is superseded by the .app bundle. - legacy = desktop / "XiaoheAgent.command" - if legacy.exists(): - legacy.unlink() - app = desktop / "XiaoheAgent.app" + _remove_legacy_launchers(desktop) + app = desktop / f"{_LAUNCHER_NAME}.app" macos = app / "Contents" / "MacOS" resources = app / "Contents" / "Resources" macos.mkdir(parents=True, exist_ok=True) @@ -264,16 +273,17 @@ def _create_launcher() -> None: else: apps = home / ".local" / "share" / "applications" apps.mkdir(parents=True, exist_ok=True) + _remove_legacy_launchers(apps) icon_line = "Icon=utilities-terminal\n" if png is not None: icon_dir = home / ".local" / "share" / "icons" / "hicolor" / "512x512" / "apps" icon_dir.mkdir(parents=True, exist_ok=True) shutil.copy2(png, icon_dir / "xiaohe-agent.png") icon_line = "Icon=xiaohe-agent\n" - (apps / "XiaoheAgent.desktop").write_text( + (apps / f"{_LAUNCHER_NAME}.desktop").write_text( "[Desktop Entry]\n" "Type=Application\n" - "Name=XiaoheAgent\n" + "Name=Xiaohe Agent\n" "Comment=Xiaohe Agent terminal\n" "Exec=xiaohe\n" "Terminal=true\n" @@ -281,7 +291,7 @@ def _create_launcher() -> None: "Categories=Utility;\n", encoding="utf-8", ) - console.print(f" [green]launcher:[/green] {apps / 'XiaoheAgent.desktop'}") + console.print(f" [green]launcher:[/green] {apps / f'{_LAUNCHER_NAME}.desktop'}") @click.command("init") diff --git a/myagents/commands/upgrade.py b/myagents/commands/upgrade.py new file mode 100644 index 0000000..042d265 --- /dev/null +++ b/myagents/commands/upgrade.py @@ -0,0 +1,238 @@ +"""``xiaohe upgrade`` — download the latest runtime package and switch to it. + +Flow: query version (from the published install.sh) -> confirm -> download +tarball with a progress bar -> install to ~/.xiaohe/runtime/ and +repoint ``current`` (keep the two newest) -> reinstall CLI tools -> sync +workspace content. Credentials come from XIAOHE_USER/XIAOHE_PASS, --user/ +--password, or an interactive prompt (never stored). +""" + +import base64 +import re +import shutil +import subprocess +import sys +import tarfile +import tempfile +import urllib.error +import urllib.request +from pathlib import Path + +import click +from rich.console import Console +from rich.progress import ( + BarColumn, + DownloadColumn, + Progress, + SpinnerColumn, + TextColumn, + TimeRemainingColumn, + TransferSpeedColumn, +) + +from myagents.commands.sync_workspace import ( + _print_report, + sync_workspace, +) +from myagents.project_root import get_workspace_root +from myagents.settings import get_setting + +stderr_console = Console(stderr=True) +console = Console() + +DEFAULT_BASE_URL = "http://1.14.226.205:8088/xiaohe" +KEEP_VERSIONS = 2 + + +def _base_url() -> str: + return str(get_setting("vps_baseurl", "") or DEFAULT_BASE_URL).rstrip("/") + + +def _runtime_root() -> Path: + return Path.home() / ".xiaohe" / "runtime" + + +def _current_version() -> str: + current = _runtime_root() / "current" + if not current.is_symlink(): + return "" + try: + return current.resolve(strict=True).name + except OSError: + return "" + + +def _auth_header(user: str, password: str) -> dict[str, str]: + token = base64.b64encode(f"{user}:{password}".encode()).decode() + return {"Authorization": f"Basic {token}"} + + +def _map_url_error(exc: Exception, url: str) -> click.ClickException: + if isinstance(exc, urllib.error.HTTPError) and exc.code in (401, 403): + return click.ClickException("Wrong account or password (server returned 401).") + if isinstance(exc, urllib.error.HTTPError): + return click.ClickException(f"Download failed: HTTP {exc.code} — {url}") + reason = getattr(exc, "reason", exc) + return click.ClickException(f"Cannot reach download server: {reason} — {url}") + + +def _fetch_text(url: str, user: str, password: str) -> str: + req = urllib.request.Request(url, headers=_auth_header(user, password)) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.read().decode("utf-8", errors="replace") + except (urllib.error.URLError, OSError) as exc: + raise _map_url_error(exc, url) from exc + + +def _download(url: str, user: str, password: str, dest: Path) -> None: + req = urllib.request.Request(url, headers=_auth_header(user, password)) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + total = int(resp.headers.get("Content-Length") or 0) or None + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + DownloadColumn(), + TransferSpeedColumn(), + TimeRemainingColumn(), + console=console, + ) as progress: + task = progress.add_task("Downloading runtime", total=total) + with dest.open("wb") as fh: + while chunk := resp.read(1 << 16): + fh.write(chunk) + progress.update(task, advance=len(chunk)) + except (urllib.error.URLError, OSError) as exc: + raise _map_url_error(exc, url) from exc + + +def _parse_version(install_sh: str) -> str | None: + match = re.search(r'^VERSION="([^"]+)"', install_sh, re.MULTILINE) + return match.group(1) if match else None + + +def _extract(tarball: Path, dest: Path) -> None: + with tarfile.open(tarball) as tf: + try: + tf.extractall(dest, filter="data") + except TypeError: # Python < 3.11.4 lacks the filter argument + tf.extractall(dest) + + +def _pip_install(pkg_dir: Path) -> subprocess.CompletedProcess: + import os + + if os.environ.get("VIRTUAL_ENV") and shutil.which("uv"): + cmd = ["uv", "pip", "install", "-e", str(pkg_dir), "--quiet"] + else: + cmd = [sys.executable, "-m", "pip", "install", "-e", str(pkg_dir), "--quiet"] + return subprocess.run(cmd, capture_output=True, text=True) + + +def _install_tools(runtime: Path) -> list[str]: + """Reinstall pip/npm tools from the new runtime. Returns warnings.""" + warnings: list[str] = [] + for pkg in ("mytoolkit", "myagents"): + with console.status(f"Installing {pkg} ..."): + result = _pip_install(runtime / "contrib" / pkg) + if result.returncode != 0: + tail = (result.stderr or "").strip().splitlines()[-1:] + warnings.append(f"{pkg} install failed: {tail[0] if tail else 'unknown error'}") + metabot = runtime / "contrib" / "metabot" + if shutil.which("npm") and (metabot / "package.json").is_file(): + steps = (["npm", "install", "--silent"], ["npm", "run", "build", "--silent"], + ["npm", "run", "update-cli", "--silent"]) + for cmd in steps: + with console.status(f"metabot: {' '.join(cmd[:2])} ..."): + result = subprocess.run(cmd, cwd=metabot, capture_output=True, text=True) + if result.returncode != 0: + warnings.append(f"metabot `{' '.join(cmd[:2])}` failed — rerun manually later") + break + elif not shutil.which("npm"): + warnings.append("npm not found — skipping metabot (xiaohe itself is unaffected)") + return warnings + + +def _prune() -> None: + versions = sorted( + (d for d in _runtime_root().iterdir() if d.is_dir() and not d.is_symlink()), + key=lambda d: d.stat().st_mtime, + reverse=True, + ) + for old in versions[KEEP_VERSIONS:]: + shutil.rmtree(old) + + +@click.command("upgrade") +@click.option("--user", envvar="XIAOHE_USER", help="Download-site account (or set XIAOHE_USER)") +@click.option("--password", envvar="XIAOHE_PASS", help="Download-site password (or set XIAOHE_PASS)") +@click.option("--force", is_flag=True, help="Reinstall even if already up to date") +def upgrade_cmd(user: str | None, password: str | None, force: bool) -> None: + """Upgrade xiaohe: fetch the latest runtime, reinstall tools, sync workspace.""" + base = _base_url() + current = _current_version() + if not current: + raise click.ClickException( + "No installed runtime found (~/.xiaohe/runtime) — run install.sh first." + ) + user = str(user or click.prompt("Download account", err=True)) + password = str(password or click.prompt("Password", hide_input=True, err=True)) + + console.print(f"[dim]Current version: {current}[/dim]") + with console.status("Checking latest version ..."): + install_sh = _fetch_text(f"{base}/install.sh", user, password) + latest = _parse_version(install_sh) + if not latest: + raise click.ClickException( + "Could not parse a version from install.sh — server content looks wrong." + ) + + if latest == current and not force: + console.print(f"[green]Already up to date ({latest}).[/green] Use --force to reinstall.") + return + if not force and not click.confirm( + f"New version {latest} (current: {current}). Upgrade now?", default=True, err=True + ): + console.print("Cancelled.") + return + + tmpdir = Path(tempfile.mkdtemp(prefix="xiaohe-upgrade-")) + try: + tarball = tmpdir / "xiaohe-agent.tar.gz" + _download(f"{base}/xiaohe-agent-latest.tar.gz", user, password, tarball) + + target = _runtime_root() / latest + with console.status(f"Installing to {target} ..."): + extract_dir = tmpdir / "extract" + _extract(tarball, extract_dir) + inner = extract_dir / "workspace" + if not (inner / "setup.sh").is_file(): + raise click.ClickException("Tarball looks wrong (setup.sh missing).") + if target.exists(): + shutil.rmtree(target) + _runtime_root().mkdir(parents=True, exist_ok=True) + shutil.move(str(inner), str(target)) + current_link = _runtime_root() / "current" + current_link.unlink(missing_ok=True) + current_link.symlink_to(target) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + warnings = _install_tools(target) + _prune() + + workspace = get_workspace_root(create=False) + if workspace.is_dir(): + report = sync_workspace(workspace) + _print_report(workspace, report) + else: + console.print(f"[yellow]workspace {workspace} missing — run 'xiaohe init' first.[/yellow]") + + for warning in warnings: + stderr_console.print(f"[yellow]warning: {warning}[/yellow]") + console.print( + f"\n[bold green]Upgraded: {current} -> {latest}[/bold green] " + "(open a new terminal; previous runtime kept for rollback)" + ) diff --git a/myagents/entrypoints.py b/myagents/entrypoints.py index 51d393b..b693b73 100644 --- a/myagents/entrypoints.py +++ b/myagents/entrypoints.py @@ -26,12 +26,14 @@ def default_agent() -> str: def build_xiaohe_cli(): - """The ``xiaohe`` command group: agent forwarding + init/sync subcommands.""" + """The ``xiaohe`` command group: agent forwarding + init/sync/upgrade.""" from myagents.commands.sync_workspace import init_cmd, sync_cmd + from myagents.commands.upgrade import upgrade_cmd xiaohe_cli = build_cli(default_agent(), prog_name="xiaohe", offer_install=True) xiaohe_cli.add_command(init_cmd) xiaohe_cli.add_command(sync_cmd) + xiaohe_cli.add_command(upgrade_cmd) return xiaohe_cli diff --git a/tests/test_sync_workspace.py b/tests/test_sync_workspace.py index ca9fe7d..97de343 100644 --- a/tests/test_sync_workspace.py +++ b/tests/test_sync_workspace.py @@ -100,19 +100,23 @@ class TestCreateLauncher: ) -> None: (runtime / "assets" / "icon").mkdir(parents=True) (runtime / "assets" / "icon" / "XiaoheAgent.icns").write_bytes(b"icns") - legacy = fake_home / "Desktop" / "XiaoheAgent.command" - legacy.write_text("#!/bin/sh\n") + legacy_cmd = fake_home / "Desktop" / "XiaoheAgent.command" + legacy_cmd.write_text("#!/bin/sh\n") + legacy_app = fake_home / "Desktop" / "XiaoheAgent.app" + legacy_app.mkdir() monkeypatch.setattr( sync_mod.os, "uname", lambda: type("U", (), {"sysname": "Darwin"}) ) sync_mod._create_launcher() - app = fake_home / "Desktop" / "XiaoheAgent.app" + app = fake_home / "Desktop" / "Xiaohe Agent.app" executable = app / "Contents" / "MacOS" / "XiaoheAgent" - assert (app / "Contents" / "Info.plist").is_file() + plist = (app / "Contents" / "Info.plist").read_text() + assert "Xiaohe Agent" in plist assert (app / "Contents" / "Resources" / "XiaoheAgent.icns").is_file() assert executable.stat().st_mode & 0o111 assert "xiaohe" in executable.read_text() - assert not legacy.exists() # superseded .command removed + assert not legacy_cmd.exists() # superseded .command removed + assert not legacy_app.exists() # renamed from XiaoheAgent.app def test_linux_desktop_entry_with_icon( self, runtime: Path, fake_home: Path, monkeypatch: pytest.MonkeyPatch @@ -124,16 +128,8 @@ class TestCreateLauncher: ) sync_mod._create_launcher() entry = ( - fake_home / ".local" / "share" / "applications" / "XiaoheAgent.desktop" + fake_home / ".local" / "share" / "applications" / "Xiaohe Agent.desktop" ) - assert "Icon=xiaohe-agent" in entry.read_text() - assert ( - fake_home - / ".local" - / "share" - / "icons" - / "hicolor" - / "512x512" - / "apps" - / "xiaohe-agent.png" - ).is_file() + text = entry.read_text() + assert "Name=Xiaohe Agent" in text + assert "Icon=xiaohe-agent" in text diff --git a/tests/test_upgrade.py b/tests/test_upgrade.py new file mode 100644 index 0000000..45810be --- /dev/null +++ b/tests/test_upgrade.py @@ -0,0 +1,134 @@ +"""Tests for myagents.commands.upgrade.""" + +import io +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from myagents.commands import upgrade as up_mod + + +@pytest.fixture() +def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) + return home + + +def _make_runtime(home: Path, version: str) -> Path: + target = home / ".xiaohe" / "runtime" / version + target.mkdir(parents=True) + current = target.parent / "current" + current.unlink(missing_ok=True) + current.symlink_to(target) + return target + + +class TestParseVersion: + def test_parses_quoted_version(self) -> None: + assert up_mod._parse_version('BASE_URL="x"\nVERSION="v1.2.3"\n') == "v1.2.3" + + def test_returns_none_on_garbage(self) -> None: + assert up_mod._parse_version("no version here") is None + + +class TestPrune: + def test_keeps_two_newest(self, fake_home: Path) -> None: + root = fake_home / ".xiaohe" / "runtime" + for name in ("v1", "v2", "v3", "v4"): + d = root / name + d.mkdir(parents=True) + up_mod._prune() + remaining = sorted(p.name for p in root.iterdir()) + assert remaining == ["v3", "v4"] + + +class TestUpgradeCommand: + def test_already_latest_shortcircuits( + self, fake_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _make_runtime(fake_home, "v1.0.0") + monkeypatch.setattr( + up_mod, "_fetch_text", lambda *a: 'VERSION="v1.0.0"\n' + ) + result = CliRunner().invoke( + up_mod.upgrade_cmd, ["--user", "u", "--password", "p"] + ) + assert result.exit_code == 0 + assert "Already up to date" in result.output + + def test_missing_runtime_errors(self, fake_home: Path) -> None: + result = CliRunner().invoke( + up_mod.upgrade_cmd, ["--user", "u", "--password", "p"] + ) + assert result.exit_code != 0 + assert "install.sh" in result.output + + def test_cancelled_by_user( + self, fake_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _make_runtime(fake_home, "v1.0.0") + monkeypatch.setattr( + up_mod, "_fetch_text", lambda *a: 'VERSION="v2.0.0"\n' + ) + result = CliRunner().invoke( + up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="n\n" + ) + assert result.exit_code == 0 + assert "Cancelled" in result.output + + +class _FakeResponse: + def __init__(self, data: bytes) -> None: + self.headers = {"Content-Length": str(len(data))} + self._buf = io.BytesIO(data) + + def read(self, n: int = -1) -> bytes: + return self._buf.read(n) + + def __enter__(self) -> "_FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + +class TestFetch: + def test_fetch_text_sends_basic_auth( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + seen: dict[str, str] = {} + + def fake_urlopen(req, timeout=0): + seen["auth"] = req.headers["Authorization"] + return _FakeResponse(b"hello") + + monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) + assert up_mod._fetch_text("http://x/", "u", "p") == "hello" + assert seen["auth"].startswith("Basic ") + + def test_401_maps_to_friendly_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + import urllib.error + + def fake_urlopen(req, timeout=0): + raise urllib.error.HTTPError("http://x/", 401, "Unauthorized", {}, None) + + monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) + with pytest.raises(Exception, match="401"): + up_mod._fetch_text("http://x/", "u", "p") + + def test_download_writes_file( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + payload = b"x" * 100_000 + monkeypatch.setattr( + "urllib.request.urlopen", + lambda req, timeout=0: _FakeResponse(payload), + ) + dest = tmp_path / "out.tar.gz" + up_mod._download("http://x/f", "u", "p", dest) + assert dest.read_bytes() == payload