From d99c5534cb3bc3964b44e3cae84e8759bd2502fb Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sat, 25 Jul 2026 12:43:24 +0800 Subject: [PATCH] refactor: drop workspace sync launcher path; simplify upgrade/switch Align with xiaohe wheel/pip install: remove sync_workspace and desktop launcher coupling from upgrade/switch flows. --- myagents/commands/info.py | 21 +- myagents/commands/switch.py | 101 +------- myagents/commands/sync_workspace.py | 375 ---------------------------- myagents/commands/uninstall.py | 17 +- myagents/commands/upgrade.py | 290 ++------------------- myagents/commands/version.py | 78 +++--- myagents/entrypoints.py | 29 ++- tests/test_switch.py | 129 +--------- tests/test_sync_workspace.py | 210 ---------------- tests/test_upgrade.py | 297 ++-------------------- tests/test_version.py | 32 +-- 11 files changed, 151 insertions(+), 1428 deletions(-) delete mode 100644 myagents/commands/sync_workspace.py delete mode 100644 tests/test_sync_workspace.py diff --git a/myagents/commands/info.py b/myagents/commands/info.py index b8e195e..31da222 100644 --- a/myagents/commands/info.py +++ b/myagents/commands/info.py @@ -1,4 +1,4 @@ -"""``xiaohe info`` — show current version, agent, and provider.""" +"""``info`` — show current version, agent, and provider (legacy myagents shim).""" from __future__ import annotations @@ -18,25 +18,25 @@ def _package_version() -> str: return "0.0.0" -def _describe_runtime() -> str: - from myagents.commands.upgrade import _current_version - - cur = _current_version() - return cur if cur else "development" +def _describe_install() -> str: + try: + return f"xiaohe-agent {version('xiaohe-agent')}" + except PackageNotFoundError: + return f"myagents {_package_version()}" @click.command("info") def info_cmd() -> None: - """Show xiaohe status: version, agent, and provider.""" - from myagents.commands.switch import _resolve_agent + """Show status: version, agent, and provider.""" from myagents.claude_settings import describe_active_backend from myagents.commands.provider import provider_list_table + from myagents.commands.switch import _resolve_agent table = Table(show_header=False, box=None, padding=(0, 1)) table.add_column("key", style="dim") table.add_column("value") - table.add_row("Version", f"myagents {_package_version()} (runtime: {_describe_runtime()})") + table.add_row("Version", _describe_install()) default_agent = _resolve_agent() table.add_row("Agent", f"{default_agent} (xiaohe → my{default_agent})") @@ -48,4 +48,5 @@ def info_cmd() -> None: provider_list_table() console.print() console.print("[dim]Switch with:[/dim]") - console.print(" xiaohe switch version xiaohe switch agent xiaohe switch provider") + console.print(" xiaohe switch agent ") + console.print(" xiaohe switch provider ") diff --git a/myagents/commands/switch.py b/myagents/commands/switch.py index 4b34a9f..dd486f4 100644 --- a/myagents/commands/switch.py +++ b/myagents/commands/switch.py @@ -1,10 +1,4 @@ -"""``xiaohe switch`` — switch version, agent, or provider. - -Subcommands: - version — switch the active runtime version (rollback/roll-forward) - agent — switch the default agent CLI (claude / kimi / codex / hermes) - provider — switch Claude Code's LLM provider (DeepSeek / Kimi / Kimi Code / Claude) -""" +"""``switch`` — agent / provider (legacy shim; product CLI is server.cli).""" from __future__ import annotations @@ -12,89 +6,11 @@ import click from rich.console import Console from myagents.entrypoints import _KNOWN_BACKENDS -from myagents.project_root import get_workspace_root from myagents.settings import get_setting, set_setting -stderr_console = Console(stderr=True) console = Console() -# ── version ──────────────────────────────────────────────────────────────── - -def _installed_versions() -> list[str]: - from myagents.commands.upgrade import _runtime_root - - root = _runtime_root() - if not root.is_dir(): - return [] - return sorted( - d.name for d in root.iterdir() if d.is_dir() and not d.is_symlink() - ) - - -@click.command("version") -@click.argument("version", required=False) -@click.option("--yes", is_flag=True, help="Skip the confirmation prompt") -def switch_version(version: str | None, yes: bool) -> None: - """Switch the active runtime to another installed version (rollback).""" - from myagents.commands.sync_workspace import _print_report, sync_workspace - from myagents.commands.upgrade import _current_version, _install_tools, _runtime_root - - versions = _installed_versions() - current = _current_version() - if not versions: - raise click.ClickException( - "No installed runtime versions found (~/.xiaohe/runtime)." - ) - - if not version: - console.print("[bold]Installed versions:[/bold]") - for name in versions: - marker = " [green](current)[/green]" if name == current else "" - console.print(f" {name}{marker}") - version = str(click.prompt("Switch to", err=True)) - - if version not in versions: - raise click.ClickException( - f"Version {version!r} is not installed. Installed: {', '.join(versions)}" - ) - if version == current: - console.print(f"Already on {version}.") - return - if not yes and not click.confirm( - f"Switch runtime {current} -> {version} and reinstall tools?", - default=True, - err=True, - ): - console.print("Cancelled.") - return - - target = _runtime_root() / version - current_link = _runtime_root() / "current" - current_link.unlink(missing_ok=True) - current_link.symlink_to(target) - - warnings = _install_tools(target) - - workspace = get_workspace_root(create=False) - if workspace.is_dir(): - try: - report = sync_workspace(workspace) - except click.ClickException as exc: - stderr_console.print(f"[yellow]workspace sync skipped: {exc.message}[/yellow]") - else: - _print_report(workspace, report) - - for warning in warnings: - stderr_console.print(f"[yellow]warning: {warning}[/yellow]") - console.print( - f"\n[bold green]Switched: {current} -> {version}[/bold green] " - "(open a new terminal to pick up the change)" - ) - - -# ── agent ────────────────────────────────────────────────────────────────── - def agent_list_table() -> None: """Print the agent list (reusable from info command).""" default = _resolve_agent() @@ -142,25 +58,18 @@ def switch_agent(agent: str | None) -> None: console.print(f"[dim]Use 'xiaohe' (or 'my{agent}') to launch.[/dim]") -# ── switch group ─────────────────────────────────────────────────────────── - @click.group("switch", invoke_without_command=True) @click.pass_context def switch_cmd(ctx: click.Context) -> None: - """Switch version, agent, or provider. - - Run without arguments to see all options. - """ + """Switch agent or provider.""" if ctx.invoked_subcommand is None: console.print("[bold]Use one of:[/bold]") - console.print(" xiaohe switch version — switch runtime version") console.print(" xiaohe switch agent — switch default agent CLI") console.print(" xiaohe switch provider — switch LLM provider") console.print() - console.print("[dim]Run 'xiaohe switch --help' for details.[/dim]") + console.print( + "[dim]Package version: xiaohe upgrade [][/dim]" + ) -switch_cmd.add_command(switch_version) switch_cmd.add_command(switch_agent) - -# provider_cmd is added in entrypoints.py (lazy import to avoid cycles) diff --git a/myagents/commands/sync_workspace.py b/myagents/commands/sync_workspace.py deleted file mode 100644 index 6e8e11d..0000000 --- a/myagents/commands/sync_workspace.py +++ /dev/null @@ -1,375 +0,0 @@ -"""``xiaohe init`` / ``xiaohe sync`` — workspace bootstrap and content sync. - -Model: the install package lives at ~/.xiaohe/runtime/ (the "runtime"); the -workspace is a user-chosen directory (settings.json workspace_root) that -receives synced skills/docs from the runtime and keeps all personal data. - -Sync is hash-based with a stamp file (/.xiaohe-sync.json): -- file only in runtime -> copy (new) -- file changed in runtime, - untouched in workspace -> overwrite (upgrade) -- file modified in workspace -> skip, report (user edit wins) -- file gone from runtime -> remove from workspace if untouched -A workspace that is a git checkout is never synced unless --force. -""" - -import hashlib -import json -import os -import shutil -from pathlib import Path - -import click -from rich.console import Console - -from myagents.project_root import get_workspace_root -from myagents.settings import get_setting, set_setting - -stderr_console = Console(stderr=True) -console = Console() - -STAMP_NAME = ".xiaohe-sync.json" - -# Root-level symlinks replicated as symlinks (targets are synced real files). -MANAGED_SYMLINKS = {"CLAUDE.md": "agents.md", ".claude": ".agents"} - -# Managed content: relative paths in the runtime tree. Directories are -# synced recursively with symlinks dereferenced (workspace must be -# self-contained — contrib/ is not synced, it is pip/npm-installed). -MANAGED_PATHS = [ - ".agents/skills", - ".agents/settings.json", - ".agents/README.md", - "agents.md", - "README.md", - "PERSONAL.md.example", - "assistant/prompts", - "assistant/pending/README.md", - "assistant/pending/CLAUDE.md", - "assistant/agent-tasks.template.md", -] - -SKELETON_DIRS = [ - "tmp", - "assistant/logs", - "assistant/checkpoints", - "assistant/knowledge", - "assistant/pending", -] - - -def get_runtime_root() -> Path | None: - """Runtime tree: $XIAOHE_RUNTIME > ~/.xiaohe/runtime/current.""" - env = os.environ.get("XIAOHE_RUNTIME") - candidates = [Path(env)] if env else [] - candidates.append(Path.home() / ".xiaohe" / "runtime" / "current") - for cand in candidates: - cand = cand.expanduser() - if (cand / "agents.md").is_file() or (cand / ".agents" / "skills").is_dir(): - return cand.resolve() - return None - - -def _runtime_version(runtime: Path) -> str: - version_file = runtime / "VERSION" - if version_file.is_file(): - return version_file.read_text(encoding="utf-8").strip() - return "" - - -def _hash_file(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as fh: - for chunk in iter(lambda: fh.read(65536), b""): - digest.update(chunk) - return digest.hexdigest() - - -def _enumerate_managed(runtime: Path) -> dict[str, str]: - """rel-path -> sha256 for every managed file in the runtime tree. - - Symlinked dirs (e.g. .agents/skills/mytoolkit -> contrib/...) are followed - and their content is hashed through the link, so the workspace receives - real self-contained files. Directory inodes are tracked to break cycles. - """ - files: dict[str, str] = {} - for rel in MANAGED_PATHS: - src = runtime / rel - if src.is_file(): - files[rel] = _hash_file(src) - continue - if not src.is_dir(): - continue - seen: set[tuple[int, int]] = set() - for dirpath, dirnames, filenames in os.walk(src, followlinks=True): - try: - real = Path(dirpath).resolve().stat() - except OSError: - dirnames[:] = [] - continue - key = (real.st_dev, real.st_ino) - if key in seen: - dirnames[:] = [] - continue - seen.add(key) - for name in sorted(filenames): - path = Path(dirpath) / name - if path.is_file(): - files[str(path.relative_to(runtime))] = _hash_file(path) - return files - - -def _load_stamp(workspace: Path) -> dict: - stamp_path = workspace / STAMP_NAME - try: - data = json.loads(stamp_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {"version": "", "files": {}} - if not isinstance(data, dict) or not isinstance(data.get("files"), dict): - return {"version": "", "files": {}} - return data - - -def _write_stamp(workspace: Path, version: str, files: dict[str, str]) -> None: - stamp = {"version": version, "files": files} - (workspace / STAMP_NAME).write_text( - json.dumps(stamp, ensure_ascii=False, indent=1) + "\n", encoding="utf-8" - ) - - -def sync_workspace(workspace: Path, force: bool = False) -> dict: - """Sync managed content from the runtime into workspace. Returns a report.""" - runtime = get_runtime_root() - if runtime is None: - raise click.ClickException( - "runtime not found (expected ~/.xiaohe/runtime/current or " - "$XIAOHE_RUNTIME) — reinstall or set XIAOHE_RUNTIME." - ) - if (workspace / ".git").exists() and not force: - raise click.ClickException( - f"{workspace} is a git checkout (dev workspace) — refusing to sync. " - "Use --force if you really mean it." - ) - - runtime_files = _enumerate_managed(runtime) - stamp = _load_stamp(workspace) - stamped: dict[str, str] = stamp.get("files", {}) - - report = {"added": [], "updated": [], "skipped": [], "removed": []} - new_stamp: dict[str, str] = {} - - for rel, runtime_hash in sorted(runtime_files.items()): - target = workspace / rel - stamped_hash = stamped.get(rel) - if stamped_hash is None: - # New managed file; do not clobber an existing divergent file. - if target.exists() and _hash_file(target) != runtime_hash: - report["skipped"].append(rel) - continue - action = "added" - else: - workspace_hash = _hash_file(target) if target.is_file() else None - if workspace_hash is not None and workspace_hash != stamped_hash: - report["skipped"].append(rel) # user-modified - new_stamp[rel] = stamped_hash - continue - if workspace_hash == runtime_hash: - new_stamp[rel] = runtime_hash - continue # already in sync - action = "updated" - target.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(runtime / rel, target) - new_stamp[rel] = runtime_hash - report[action].append(rel) - - for rel, stamped_hash in sorted(stamped.items()): - if rel in runtime_files: - continue - target = workspace / rel - if target.is_file() and _hash_file(target) == stamped_hash: - target.unlink() - report["removed"].append(rel) - elif target.exists(): - report["skipped"].append(rel) - new_stamp[rel] = stamped_hash - - for link, target_rel in MANAGED_SYMLINKS.items(): - link_path = workspace / link - # Never clobber: an existing entry (real file, dir, or link) stays put. - if link_path.is_symlink() or link_path.exists(): - continue - if (workspace / target_rel).exists(): - link_path.symlink_to(target_rel) - - _write_stamp(workspace, _runtime_version(runtime), new_stamp) - return report - - -def _print_report(workspace: Path, report: dict) -> None: - console.print(f"[bold]Synced into[/bold] {workspace}") - for action, label in (("added", "green"), ("updated", "cyan")): - if report[action]: - console.print(f" [{label}]{action}: {len(report[action])}[/{label}]") - if report["removed"]: - console.print(f" [yellow]removed: {len(report['removed'])}[/yellow]") - if report["skipped"]: - console.print( - f" [yellow]skipped (locally modified): {len(report['skipped'])}[/yellow]" - ) - for rel in report["skipped"][:10]: - console.print(f" [dim]- {rel}[/dim]") - - -_LAUNCHER_NAME = "Xiaohe Agent" -_LEGACY_LAUNCHER_NAMES = ("XiaoheAgent.command", "XiaoheAgent.app", "XiaoheAgent.desktop") - -_APP_INFO_PLIST = """ - - - - CFBundleNameXiaohe Agent - CFBundleDisplayNameXiaohe Agent - CFBundleIdentifiercom.xiaohe.agent - CFBundleVersion1 - CFBundlePackageTypeAPPL - CFBundleExecutableXiaoheAgent - CFBundleIconFileXiaoheAgent - LSMinimumSystemVersion11.0 - NSHighResolutionCapable - - -""" - -_APP_EXECUTABLE = """#!/usr/bin/env bash -# 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' \\ - -e 'tell application "Terminal" to do script "xiaohe"' -else - osascript -e 'display dialog "xiaohe command not found — run install.sh first" buttons {"OK"}' -fi -""" - - -def _icon_assets() -> tuple[Path | None, Path | None]: - """(icns, png) shipped in the runtime tree, if present.""" - runtime = get_runtime_root() - if runtime is None: - return None, None - icons = runtime / "assets" / "icon" - icns = icons / "XiaoheAgent.icns" - png = icons / "xiaohe-icon-512.png" - 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() - icns, png = _icon_assets() - if os.uname().sysname == "Darwin": # noqa: PLR2004 — platform check - desktop = home / "Desktop" - if not desktop.is_dir(): - return - _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) - resources.mkdir(parents=True, exist_ok=True) - (app / "Contents" / "Info.plist").write_text(_APP_INFO_PLIST, encoding="utf-8") - executable = macos / "XiaoheAgent" - executable.write_text(_APP_EXECUTABLE, encoding="utf-8") - executable.chmod(0o755) - if icns is not None: - shutil.copy2(icns, resources / "XiaoheAgent.icns") - console.print(f" [green]launcher:[/green] {app}") - 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 / f"{_LAUNCHER_NAME}.desktop").write_text( - "[Desktop Entry]\n" - "Type=Application\n" - "Name=Xiaohe Agent\n" - "Comment=Xiaohe Agent terminal\n" - "Exec=xiaohe\n" - "Terminal=true\n" - f"{icon_line}" - "Categories=Utility;\n", - encoding="utf-8", - ) - console.print(f" [green]launcher:[/green] {apps / f'{_LAUNCHER_NAME}.desktop'}") - - -@click.command("init") -@click.argument("path", required=False) -@click.option("--force", is_flag=True, help="Sync even into a git checkout.") -def init_cmd(path: str | None, force: bool) -> None: - """First-run workspace setup: choose path, lay skeleton, sync content.""" - configured = str(get_setting("workspace_root", "") or "").strip() - if path: - workspace = Path(path).expanduser().resolve() - elif configured: - workspace = Path(configured).expanduser().resolve() - else: - default = str(Path.home() / "workspace") - answer = click.prompt( - "Workspace path", default=default, show_default=True, err=True - ) - workspace = Path(answer).expanduser().resolve() - - if not configured or configured != str(workspace): - set_setting("workspace_root", str(workspace)) - console.print(f"[green]workspace_root saved to settings.json:[/green] {workspace}") - - workspace.mkdir(parents=True, exist_ok=True) - for rel in SKELETON_DIRS: - (workspace / rel).mkdir(parents=True, exist_ok=True) - - if get_runtime_root() is not None: - report = sync_workspace(workspace, force=force) - _print_report(workspace, report) - else: - stderr_console.print( - "[yellow]runtime not found — skeleton only; " - "skills/docs will sync after reinstall.[/yellow]" - ) - - # Worklog task board: instantiate the live file from the synced template. - template = workspace / "assistant" / "agent-tasks.template.md" - live = workspace / "assistant" / "agent-tasks.md" - if template.is_file() and not live.exists(): - shutil.copy2(template, live) - console.print(" [green]created:[/green] assistant/agent-tasks.md (from template)") - - _create_launcher() - console.print("\n[bold green]Done.[/bold green] Daily use: [cyan]xiaohe[/cyan]") - - -@click.command("sync") -@click.option("--force", is_flag=True, help="Sync even into a git checkout.") -def sync_cmd(force: bool) -> None: - """Sync skills/docs from the installed runtime into the workspace.""" - workspace = get_workspace_root(create=False) - if not workspace.is_dir(): - raise click.ClickException( - f"workspace {workspace} does not exist — run 'xiaohe init' first." - ) - report = sync_workspace(workspace, force=force) - _print_report(workspace, report) diff --git a/myagents/commands/uninstall.py b/myagents/commands/uninstall.py index 515aa45..d16b0e9 100644 --- a/myagents/commands/uninstall.py +++ b/myagents/commands/uninstall.py @@ -1,10 +1,7 @@ -"""``xiaohe uninstall`` — remove the CLI layer, keep runtime/config/workspace. +"""``uninstall`` — remove CLI entry points; keep config/workspace by default. -Default: removes the entry points (xiaohe/myclaude/.../mytoolkit), metabot -CLI, shell completions, and the desktop launcher. Keeps ~/.xiaohe (runtime, -settings, keys), ~/.metabot, ~/.mytoolkit and the workspace. ``--all`` also -deletes ~/.xiaohe, ~/.metabot and ~/.mytoolkit (every saved credential); -the workspace is never touched. +Also removes leftover metabot stubs if present. ``--all`` deletes ~/.xiaohe, +~/.metabot and ~/.mytoolkit; workspace is never touched. """ import os @@ -78,7 +75,7 @@ def _pip_uninstall() -> list[str]: @click.command("uninstall") @click.option("--all", "remove_all", is_flag=True, - help="Also remove ~/.xiaohe (runtime, settings, API keys)") + help="Also remove ~/.xiaohe (settings, API keys)") @click.option("--yes", is_flag=True, help="Skip the first confirmation (--all still requires the full chain)") def uninstall_cmd(remove_all: bool, yes: bool) -> None: @@ -95,12 +92,12 @@ def uninstall_cmd(remove_all: bool, yes: bool) -> None: console.print(" - desktop launcher (Xiaohe Agent.app / Xiaohe Agent.desktop)") console.print(" - pip packages: mytoolkit, myagents") if remove_all: - console.print(" - [red]~/.xiaohe (runtime, settings.json, config.json keys)[/red]") + console.print(" - [red]~/.xiaohe (settings, secrets, keys)[/red]") console.print(" - [red]~/.metabot (Feishu bot config)[/red]") console.print(" - [red]~/.mytoolkit (mytoolkit config, incl. keys)[/red]") console.print("[bold]Will keep:[/bold]") if not remove_all: - console.print(" - ~/.xiaohe (runtime, settings, keys)") + console.print(" - ~/.xiaohe (settings, keys)") console.print(" - ~/.metabot and ~/.mytoolkit (bot/tool configs)") console.print(f" - workspace: {get_workspace_root(create=False)}") console.print(" - rc-file edits (PATH line, ANTHROPIC_* exports) and the claude CLI") @@ -112,7 +109,7 @@ def uninstall_cmd(remove_all: bool, yes: bool) -> None: # Deleting these dirs destroys settings and every saved credential — # always require the full confirmation chain, even with --yes. if not click.confirm( - "This also deletes ~/.xiaohe, ~/.metabot and ~/.mytoolkit — runtime, " + "This also deletes ~/.xiaohe, ~/.metabot and ~/.mytoolkit — config, " "bot config and ALL saved API keys. Continue?", default=False, err=True, diff --git a/myagents/commands/upgrade.py b/myagents/commands/upgrade.py index 225e6b1..9110172 100644 --- a/myagents/commands/upgrade.py +++ b/myagents/commands/upgrade.py @@ -1,277 +1,37 @@ -"""``xiaohe upgrade`` — download the latest runtime package and switch to it. +"""``upgrade`` — forward to ``xiaohe upgrade`` (pip wheel). -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). +Old runtime-tarball install under ``~/.xiaohe/runtime`` is removed. """ -import base64 -import hashlib -import re +from __future__ import annotations + 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-agent" -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 _sha256(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as fh: - for chunk in iter(lambda: fh.read(1 << 16), b""): - digest.update(chunk) - return digest.hexdigest() - - -def _verify_checksum( - tarball: Path, sha256_text: str, url: str -) -> None: - """Verify tarball against the published " " line.""" - expected = sha256_text.split()[0].strip().lower() if sha256_text.split() else "" - if not expected or not all(c in "0123456789abcdef" for c in expected): - raise click.ClickException(f"Checksum file looks wrong — {url}") - actual = _sha256(tarball) - if actual != expected: - raise click.ClickException( - "Checksum mismatch — the download is corrupted or tampered with. " - "Aborting; nothing was installed." - ) - - -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 - import sysconfig - - 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"] - # Ubuntu 24.04+ marks the system Python externally-managed (PEP 668); - # without the flag pip refuses to install there at all. - stdlib = Path(sysconfig.get_path("stdlib")) - if (stdlib / "EXTERNALLY-MANAGED").exists(): - cmd.append("--break-system-packages") - cmd += ["-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: +@click.argument("version", required=False) +@click.option("--force", is_flag=True, help="Reinstall even if already on that version") +@click.option("--user", "user", default=None, hidden=True) +@click.option("--password", default=None, hidden=True) +def upgrade_cmd( + version: str | None, + force: bool, + user: str | None, + password: str | None, +) -> None: + """Forward to ``xiaohe upgrade`` (wheel).""" + del user, password + xiaohe = shutil.which("xiaohe") + if not xiaohe: raise click.ClickException( - "No installed runtime found (~/.xiaohe/runtime) — run install.sh first." + "Install xiaohe-agent, then run: xiaohe upgrade" ) - 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-latest.tar.gz" - _download(f"{base}/xiaohe-agent-latest.tar.gz", user, password, tarball) - with console.status("Verifying checksum ..."): - sha_url = f"{base}/xiaohe-agent-latest.tar.gz.sha256" - _verify_checksum(tarball, _fetch_text(sha_url, user, password), sha_url) - - 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(): - try: - report = sync_workspace(workspace) - except click.ClickException as exc: - # e.g. dev machine: workspace is a git checkout — not a failure. - stderr_console.print(f"[yellow]workspace sync skipped: {exc.message}[/yellow]") - else: - _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; roll back anytime with 'xiaohe switch')" - ) + cmd = [xiaohe, "upgrade"] + if version: + cmd.append(version) + if force: + cmd.append("--force") + raise SystemExit(subprocess.call(cmd)) diff --git a/myagents/commands/version.py b/myagents/commands/version.py index 1ba369d..f93afc5 100644 --- a/myagents/commands/version.py +++ b/myagents/commands/version.py @@ -1,7 +1,8 @@ -"""``xiaohe version`` — show whether the CLI runs a dev checkout or an installed runtime.""" +"""``version`` — show package install location (legacy myagents shim).""" import shutil import subprocess +from importlib.metadata import PackageNotFoundError, version from pathlib import Path import click @@ -14,11 +15,15 @@ def _git_describe(tree: Path) -> str: try: describe = subprocess.run( ["git", "-C", str(tree), "describe", "--tags", "--dirty", "--always"], - capture_output=True, text=True, timeout=5, + capture_output=True, + text=True, + timeout=5, ) branch = subprocess.run( ["git", "-C", str(tree), "branch", "--show-current"], - capture_output=True, text=True, timeout=5, + capture_output=True, + text=True, + timeout=5, ) except (OSError, subprocess.TimeoutExpired): return "unknown" @@ -30,57 +35,58 @@ def _git_describe(tree: Path) -> str: def describe_tree(tree: Path) -> tuple[str, str]: - """Classify a package source tree: ('installed'|'development'|'unknown', detail).""" + """Classify a package source tree.""" parts = tree.parts if ".xiaohe" in parts and "runtime" in parts: + # Legacy tarball layout (no longer installed by upgrade). try: idx = parts.index("runtime") - return "installed", parts[idx + 1] + return "legacy-runtime", parts[idx + 1] except (ValueError, IndexError): - return "installed", "unknown" - if (tree / ".git").exists(): + return "legacy-runtime", "unknown" + if (tree / ".git").exists() or (tree.parent / ".git").exists(): + # contrib/myagents → repo root may hold .git + probe = tree + for _ in range(3): + if (probe / ".git").exists(): + return "development", _git_describe(probe) + probe = probe.parent return "development", _git_describe(tree) return "unknown", "" def _pkg_tree(module_file: str) -> Path: - # /myagents/__init__.py -> return Path(module_file).resolve().parents[1] -def _report_pkg(label: str, module_file: str) -> str: - tree = _pkg_tree(module_file) - mode, detail = describe_tree(tree) - line = { - "installed": f"installed runtime [cyan]{detail}[/cyan]", - "development": f"[green]development checkout[/green] — {detail}", - }.get(mode, "unknown") - console.print(f" {label}: {line}") - console.print(f" [dim]{tree}[/dim]") - return mode - - @click.command("version") def version_cmd() -> None: - """Show which tree the CLI runs from: dev checkout or installed runtime.""" + """Show which tree the CLI runs from.""" import myagents - bin_path = shutil.which("xiaohe") or "?" - console.print(f"[bold]xiaohe[/bold] (bin: {bin_path})") - mode = _report_pkg("myagents", myagents.__file__) + bin_path = shutil.which("xiaohe") or shutil.which("myagents") or "?" + console.print(f"[bold]myagents[/bold] (bin: {bin_path})") + try: - import mytoolkit + console.print(f" xiaohe-agent: [cyan]{version('xiaohe-agent')}[/cyan]") + except PackageNotFoundError: + pass - _report_pkg("mytoolkit", mytoolkit.__file__) - except ImportError: - console.print(" mytoolkit: [yellow]not importable[/yellow]") - - current = Path.home() / ".xiaohe" / "runtime" / "current" - if current.is_symlink(): - console.print(f"runtime snapshot: [cyan]{current.resolve().name}[/cyan] [dim]({current.parent})[/dim]") - - if mode == "installed": + tree = _pkg_tree(myagents.__file__) + mode, detail = describe_tree(tree) + if mode == "development": + console.print(f" myagents: [green]development[/green] — {detail}") + elif mode == "legacy-runtime": console.print( - "\n[dim]Dev machine? Point the bins at your repo instead:[/dim]\n" - " pip install -e /contrib/mytoolkit -e /contrib/myagents" + f" myagents: [yellow]legacy runtime {detail}[/yellow] " + "(use xiaohe upgrade / pip install)" + ) + else: + console.print(f" myagents: installed [dim]{tree}[/dim]") + + legacy = Path.home() / ".xiaohe" / "runtime" / "current" + if legacy.is_symlink(): + console.print( + f" [dim]leftover ~/.xiaohe/runtime → {legacy.resolve().name} " + "(safe to remove after wheel install)[/dim]" ) diff --git a/myagents/entrypoints.py b/myagents/entrypoints.py index dba972c..4c062b2 100644 --- a/myagents/entrypoints.py +++ b/myagents/entrypoints.py @@ -2,6 +2,8 @@ import re +import click + from myagents.launcher import build_cli claude_cli = build_cli("claude", prog_name="myclaude") @@ -26,16 +28,32 @@ def default_agent() -> str: def build_xiaohe_cli(): - """The ``xiaohe`` command group: agent forwarding + info/init/sync/upgrade/switch/uninstall/version.""" + """Legacy ``xiaohe`` group for myagents-only installs. + + The product entrypoint is ``server.cli:main`` (xiaohe-agent wheel). + """ from myagents.commands.info import info_cmd from myagents.commands.provider import provider_cmd from myagents.commands.switch import switch_cmd - from myagents.commands.sync_workspace import init_cmd, sync_cmd from myagents.commands.uninstall import uninstall_cmd from myagents.commands.upgrade import upgrade_cmd from myagents.commands.version import version_cmd - switch_cmd.add_command(provider_cmd) + @click.command("init") + def init_cmd() -> None: + """Removed — use ``xiaohe workspace init``.""" + raise click.ClickException("Use: xiaohe workspace init") + + @click.command("sync") + def sync_cmd() -> None: + """Removed — workspace is git-managed; product skills use ``make sync``.""" + raise click.ClickException( + "Workspace content is git-managed. " + "Product skills: make sync (in xiaohe-agent)." + ) + + if "provider" not in switch_cmd.commands: + switch_cmd.add_command(provider_cmd) xiaohe_cli = build_cli(default_agent(), prog_name="xiaohe", offer_install=True) xiaohe_cli.add_command(info_cmd) @@ -102,10 +120,9 @@ def hermes_main() -> None: def xiaohe_main() -> None: - """Run ``xiaohe``: default-agent launcher plus init/sync subcommands. + """Legacy ``xiaohe`` entry (myagents-only installs). - Same options as the corresponding ``my`` command; when the - underlying CLI is missing, offers to install it interactively. + Prefer the xiaohe-agent wheel entrypoint ``server.cli:main``. """ from myagents.commands.completion_install import ( ensure_completions_installed, diff --git a/tests/test_switch.py b/tests/test_switch.py index 765a054..7b23a5e 100644 --- a/tests/test_switch.py +++ b/tests/test_switch.py @@ -1,134 +1,13 @@ -"""Tests for myagents.commands.switch (version / agent subcommands).""" +"""Tests for myagents.commands.switch (legacy shim).""" -from pathlib import Path - -import pytest from click.testing import CliRunner from myagents.commands import switch as sw_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)) - workspace = tmp_path / "ws" - workspace.mkdir() - monkeypatch.setattr(sw_mod, "get_workspace_root", lambda create=False: workspace) - # sync_workspace is imported locally in switch_version - import myagents.commands.sync_workspace as sync_mod - - monkeypatch.setattr( - sync_mod, - "sync_workspace", - lambda _ws: {"added": [], "updated": [], "skipped": [], "removed": []}, - ) - # _install_tools is imported locally inside switch_version from upgrade - import myagents.commands.upgrade as ug_mod - - monkeypatch.setattr(ug_mod, "_install_tools", lambda _rt: []) - return home - - -def _make_versions(home: Path, versions: tuple[str, ...], current: str) -> None: - root = home / ".xiaohe" / "runtime" - for name in versions: - (root / name).mkdir(parents=True) - (root / "current").symlink_to(root / current) - - -class TestSwitchVersion: - def test_switches_current_and_reinstalls( - self, fake_home: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - _make_versions(fake_home, ("v1", "v2"), "v2") - installed: list[str] = [] - import myagents.commands.upgrade as ug_mod - - monkeypatch.setattr( - ug_mod, - "_install_tools", - lambda _rt: (installed.append(_rt.name), [])[1] or [], - ) - result = CliRunner().invoke(sw_mod.switch_version, ["v1", "--yes"]) - assert result.exit_code == 0, result.output - assert "Switched:" in result.output - current = fake_home / ".xiaohe" / "runtime" / "current" - assert current.resolve().name == "v1" - assert installed == ["v1"] - - def test_unknown_version_errors(self, fake_home: Path) -> None: - _make_versions(fake_home, ("v1", "v2"), "v2") - result = CliRunner().invoke(sw_mod.switch_version, ["v9", "--yes"]) - assert result.exit_code != 0 - assert "not installed" in result.output - - def test_already_current_is_noop(self, fake_home: Path) -> None: - _make_versions(fake_home, ("v1", "v2"), "v2") - result = CliRunner().invoke(sw_mod.switch_version, ["v2", "--yes"]) - assert result.exit_code == 0 - assert "Already on v2" in result.output - - def test_interactive_list_and_prompt(self, fake_home: Path) -> None: - _make_versions(fake_home, ("v1", "v2"), "v2") - result = CliRunner().invoke(sw_mod.switch_version, [], input="v2\ny\n") - assert result.exit_code == 0, result.output - assert "v2" in result.output and "(current)" in result.output - - def test_cancelled_keeps_current(self, fake_home: Path) -> None: - _make_versions(fake_home, ("v1", "v2"), "v2") - result = CliRunner().invoke(sw_mod.switch_version, ["v1"], input="n\n") - assert result.exit_code == 0 - assert "Cancelled" in result.output - assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v2" - - def test_no_versions_errors(self, fake_home: Path) -> None: - result = CliRunner().invoke(sw_mod.switch_version, ["v1", "--yes"]) - assert result.exit_code != 0 - assert "No installed runtime" in result.output - - -class TestSwitchAgent: - def test_bare_invocation_lists_agents(self) -> None: - result = CliRunner().invoke( - sw_mod.switch_agent, [], input="claude\n" - ) - assert result.exit_code == 0, result.output - assert "Available agents" in result.output - assert "claude" in result.output - assert "kimi" in result.output - - def test_switch_to_kimi(self) -> None: - result = CliRunner().invoke(sw_mod.switch_agent, ["kimi"]) - assert result.exit_code == 0, result.output - assert "Switched" in result.output - - def test_already_current_noop(self) -> None: - from myagents.settings import get_setting - - current = get_setting("default_agent", "claude") - result = CliRunner().invoke(sw_mod.switch_agent, [current]) - assert result.exit_code == 0 - assert "Already on" in result.output - - def test_unknown_agent_errors(self) -> None: - result = CliRunner().invoke(sw_mod.switch_agent, ["unknown-ai"]) - assert result.exit_code != 0 - assert "Unknown agent" in result.output - - -class TestSwitchGroup: - def test_bare_invocation_shows_options(self) -> None: +class TestSwitchHelp: + def test_group_mentions_upgrade(self) -> None: result = CliRunner().invoke(sw_mod.switch_cmd, []) assert result.exit_code == 0 - assert "switch version" in result.output - assert "switch agent" in result.output - assert "switch provider" in result.output - - def test_help_lists_subcommands(self) -> None: - result = CliRunner().invoke(sw_mod.switch_cmd, ["--help"]) - assert result.exit_code == 0 - assert "version" in result.output + assert "xiaohe upgrade" in result.output assert "agent" in result.output - assert "provider" in result.output diff --git a/tests/test_sync_workspace.py b/tests/test_sync_workspace.py deleted file mode 100644 index 597bf4e..0000000 --- a/tests/test_sync_workspace.py +++ /dev/null @@ -1,210 +0,0 @@ -"""Tests for myagents.commands.sync_workspace.""" - -import json -from pathlib import Path - -import click -import pytest - -from myagents.commands import sync_workspace as sync_mod - - -@pytest.fixture() -def runtime(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - rt = tmp_path / "runtime" - (rt / ".agents" / "skills" / "demo").mkdir(parents=True) - (rt / ".agents" / "skills" / "demo" / "SKILL.md").write_text("demo v1") - (rt / "agents.md").write_text("agent rules v1") - (rt / "VERSION").write_text("v0.0.0-test\n") - monkeypatch.setattr(sync_mod, "get_runtime_root", lambda: rt) - return rt - - -@pytest.fixture() -def workspace(tmp_path: Path) -> Path: - ws = tmp_path / "workspace" - ws.mkdir() - return ws - - -def _read_stamp(ws: Path) -> dict: - return json.loads((ws / sync_mod.STAMP_NAME).read_text()) - - -class TestSyncWorkspace: - def test_new_files_copied_and_symlinks_made( - self, runtime: Path, workspace: Path - ) -> None: - report = sync_mod.sync_workspace(workspace) - skill = workspace / ".agents" / "skills" / "demo" / "SKILL.md" - assert skill.read_text() == "demo v1" - assert (workspace / "agents.md").read_text() == "agent rules v1" - assert (workspace / "CLAUDE.md").is_symlink() - assert (workspace / ".claude").is_symlink() - assert report["added"] and not report["skipped"] - assert _read_stamp(workspace)["version"] == "v0.0.0-test" - - def test_second_run_is_noop(self, runtime: Path, workspace: Path) -> None: - sync_mod.sync_workspace(workspace) - report = sync_mod.sync_workspace(workspace) - assert report == {"added": [], "updated": [], "skipped": [], "removed": []} - - def test_upgrade_overwrites_untouched_file( - self, runtime: Path, workspace: Path - ) -> None: - sync_mod.sync_workspace(workspace) - (runtime / "agents.md").write_text("agent rules v2") - report = sync_mod.sync_workspace(workspace) - assert (workspace / "agents.md").read_text() == "agent rules v2" - assert "agents.md" in report["updated"] - - def test_user_modified_file_is_skipped( - self, runtime: Path, workspace: Path - ) -> None: - sync_mod.sync_workspace(workspace) - (workspace / "agents.md").write_text("my local edits") - (runtime / "agents.md").write_text("agent rules v2") - report = sync_mod.sync_workspace(workspace) - assert (workspace / "agents.md").read_text() == "my local edits" - assert "agents.md" in report["skipped"] - - def test_removed_from_runtime_is_removed_from_workspace( - self, runtime: Path, workspace: Path - ) -> None: - sync_mod.sync_workspace(workspace) - (runtime / ".agents" / "skills" / "demo" / "SKILL.md").unlink() - report = sync_mod.sync_workspace(workspace) - assert not (workspace / ".agents" / "skills" / "demo" / "SKILL.md").exists() - assert report["removed"] - - def test_git_checkout_refused_without_force( - self, runtime: Path, workspace: Path - ) -> None: - (workspace / ".git").mkdir() - with pytest.raises(click.ClickException): - sync_mod.sync_workspace(workspace) - report = sync_mod.sync_workspace(workspace, force=True) - assert report["added"] - - def test_symlinked_skill_dir_is_dereferenced( - self, runtime: Path, workspace: Path - ) -> None: - # Skills living in contrib/ are linked into .agents/skills (relative - # symlink); the workspace must receive real self-contained files. - real = runtime / "contrib" / "mytoolkit" / ".claude" / "skills" / "mytoolkit" - real.mkdir(parents=True) - (real / "SKILL.md").write_text("toolkit skill v1") - import os - - rel_target = os.path.relpath(real, runtime / ".agents" / "skills") - (runtime / ".agents" / "skills" / "mytoolkit").symlink_to( - rel_target, target_is_directory=True - ) - report = sync_mod.sync_workspace(workspace) - copied = workspace / ".agents" / "skills" / "mytoolkit" / "SKILL.md" - assert copied.is_file() and not copied.is_symlink() - assert copied.read_text() == "toolkit skill v1" - assert ".agents/skills/mytoolkit/SKILL.md" in report["added"] - # Second run: content identical -> no spurious updates. - assert sync_mod.sync_workspace(workspace) == { - "added": [], "updated": [], "skipped": [], "removed": [] - } - - def test_existing_real_managed_symlink_target_is_kept( - self, runtime: Path, workspace: Path - ) -> None: - # A pre-existing real CLAUDE.md must not be replaced by a symlink. - (workspace / "CLAUDE.md").write_text("my own claude rules") - sync_mod.sync_workspace(workspace) - claude_md = workspace / "CLAUDE.md" - assert not claude_md.is_symlink() - assert claude_md.read_text() == "my own claude rules" - assert (workspace / ".claude").is_symlink() # other links still made - - -class TestCreateLauncher: - @pytest.fixture() - def fake_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - home = tmp_path / "home" - (home / "Desktop").mkdir(parents=True) - monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) - return home - - def test_macos_app_bundle_with_icon( - self, runtime: Path, fake_home: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - (runtime / "assets" / "icon").mkdir(parents=True) - (runtime / "assets" / "icon" / "XiaoheAgent.icns").write_bytes(b"icns") - 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" / "Xiaohe Agent.app" - executable = app / "Contents" / "MacOS" / "XiaoheAgent" - 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_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 - ) -> None: - (runtime / "assets" / "icon").mkdir(parents=True) - (runtime / "assets" / "icon" / "xiaohe-icon-512.png").write_bytes(b"png") - monkeypatch.setattr( - sync_mod.os, "uname", lambda: type("U", (), {"sysname": "Linux"}) - ) - sync_mod._create_launcher() - entry = ( - fake_home / ".local" / "share" / "applications" / "Xiaohe Agent.desktop" - ) - text = entry.read_text() - assert "Name=Xiaohe Agent" in text - assert "Icon=xiaohe-agent" in text - - -class TestInitCommand: - @pytest.fixture() - def fake_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - home = tmp_path / "home" - home.mkdir() - monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) - monkeypatch.setattr(sync_mod, "get_setting", lambda *a, **kw: "") - monkeypatch.setattr(sync_mod, "set_setting", lambda *a, **kw: None) - return home - - def test_agent_tasks_created_from_template( - self, runtime: Path, workspace: Path, fake_home: Path - ) -> None: - from click.testing import CliRunner - - template = runtime / "assistant" / "agent-tasks.template.md" - template.parent.mkdir(parents=True) - template.write_text("# task board template") - result = CliRunner().invoke(sync_mod.init_cmd, [str(workspace)]) - assert result.exit_code == 0, result.output - live = workspace / "assistant" / "agent-tasks.md" - assert live.read_text() == "# task board template" - assert (workspace / "tmp").is_dir() # skeleton dirs created - - def test_existing_agent_tasks_untouched( - self, runtime: Path, workspace: Path, fake_home: Path - ) -> None: - from click.testing import CliRunner - - template = runtime / "assistant" / "agent-tasks.template.md" - template.parent.mkdir(parents=True) - template.write_text("# task board template") - live = workspace / "assistant" / "agent-tasks.md" - live.parent.mkdir(parents=True) - live.write_text("# my live board") - result = CliRunner().invoke(sync_mod.init_cmd, [str(workspace)]) - assert result.exit_code == 0, result.output - assert live.read_text() == "# my live board" diff --git a/tests/test_upgrade.py b/tests/test_upgrade.py index 21c4889..75dee39 100644 --- a/tests/test_upgrade.py +++ b/tests/test_upgrade.py @@ -1,289 +1,26 @@ -"""Tests for myagents.commands.upgrade.""" +"""Tests for myagents.commands.upgrade (forwards to xiaohe upgrade).""" -import io -from pathlib import Path +from unittest.mock import patch -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_forwards_to_xiaohe(self) -> None: + with ( + patch.object(up_mod.shutil, "which", return_value="/bin/xiaohe"), + patch.object(up_mod.subprocess, "call", return_value=0) as call, + ): + result = CliRunner().invoke(up_mod.upgrade_cmd, ["0.5.1", "--force"]) + assert result.exit_code == 0 + call.assert_called_once_with( + ["/bin/xiaohe", "upgrade", "0.5.1", "--force"] + ) - 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 - - -def _make_tarball_bytes() -> bytes: - import tarfile - - buf = io.BytesIO() - with tarfile.open(fileobj=buf, mode="w:gz") as tf: - data = b"#!/bin/sh\n" - info = tarfile.TarInfo("workspace/setup.sh") - info.size = len(data) - tf.addfile(info, io.BytesIO(data)) - return buf.getvalue() - - -def _wire_upgrade( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - payload: bytes, - sha: str, -) -> None: - import hashlib - - def fake_fetch(url: str, user: str, password: str) -> str: - if url.endswith("install.sh"): - return 'VERSION="v2.0.0"\n' - if url.endswith(".sha256"): - return f"{sha} xiaohe-agent-latest.tar.gz\n" - raise AssertionError(f"unexpected url {url}") - - monkeypatch.setattr(up_mod, "_fetch_text", fake_fetch) - monkeypatch.setattr( - up_mod, "_download", lambda url, u, p, dest: dest.write_bytes(payload) - ) - monkeypatch.setattr(up_mod, "_install_tools", lambda runtime: []) - workspace = tmp_path / "ws" - workspace.mkdir() - monkeypatch.setattr(up_mod, "get_workspace_root", lambda create=False: workspace) - monkeypatch.setattr( - up_mod, - "sync_workspace", - lambda ws: {"added": [], "updated": [], "skipped": [], "removed": []}, - ) - - -class TestUpgradeFlow: - def test_downloads_verifies_and_switches( - self, fake_home: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path - ) -> None: - import hashlib - - _make_runtime(fake_home, "v1.0.0") - payload = _make_tarball_bytes() - sha = hashlib.sha256(payload).hexdigest() - _wire_upgrade(monkeypatch, tmp_path, payload, sha) - result = CliRunner().invoke( - up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="y\n" - ) - assert result.exit_code == 0, result.output - assert "Upgraded: v1.0.0 -> v2.0.0" in result.output - current = fake_home / ".xiaohe" / "runtime" / "current" - assert current.resolve().name == "v2.0.0" - assert (current / "setup.sh").is_file() - - def test_checksum_mismatch_aborts_before_install( - self, fake_home: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path - ) -> None: - _make_runtime(fake_home, "v1.0.0") - payload = _make_tarball_bytes() - _wire_upgrade(monkeypatch, tmp_path, payload, "0" * 64) - result = CliRunner().invoke( - up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="y\n" - ) - assert result.exit_code != 0 - assert "Checksum mismatch" in result.output - # Old runtime untouched, no new version dir. - current = fake_home / ".xiaohe" / "runtime" / "current" - assert current.resolve().name == "v1.0.0" - assert not (fake_home / ".xiaohe" / "runtime" / "v2.0.0").exists() - - -class TestVerifyChecksum: - def test_accepts_matching(self, tmp_path: Path) -> None: - import hashlib - - blob = tmp_path / "f.tar.gz" - blob.write_bytes(b"data") - sha = hashlib.sha256(b"data").hexdigest() - up_mod._verify_checksum(blob, f"{sha} f.tar.gz\n", "http://x/") - - def test_rejects_mismatch_and_garbage(self, tmp_path: Path) -> None: - import click - - blob = tmp_path / "f.tar.gz" - blob.write_bytes(b"data") - with pytest.raises(click.ClickException, match="Checksum mismatch"): - up_mod._verify_checksum(blob, f"{'0' * 64} f.tar.gz\n", "http://x/") - with pytest.raises(click.ClickException, match="looks wrong"): - up_mod._verify_checksum(blob, "not-a-checksum\n", "http://x/") - - -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 - - -class TestPipInstallFlags: - def test_uses_break_system_packages_on_externally_managed( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - import sysconfig - - # Force the plain-pip branch regardless of how pytest was launched - # (e.g. `uv run` sets VIRTUAL_ENV, which would select the uv branch). - monkeypatch.delenv("VIRTUAL_ENV", raising=False) - em = tmp_path / "EXTERNALLY-MANAGED" - em.write_text("[externally-managed]\n") - monkeypatch.setattr( - sysconfig, "get_path", lambda name: str(tmp_path) if name == "stdlib" else "" - ) - seen: list[list[str]] = [] - monkeypatch.setattr( - "subprocess.run", - lambda cmd, **kw: seen.append(cmd) - or type("R", (), {"returncode": 0, "stderr": ""})(), - ) - up_mod._pip_install(tmp_path / "pkg") - assert "--break-system-packages" in seen[0] - - def test_skips_flag_without_externally_managed( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - import sysconfig - - monkeypatch.delenv("VIRTUAL_ENV", raising=False) - monkeypatch.setattr( - sysconfig, "get_path", lambda name: str(tmp_path) if name == "stdlib" else "" - ) - seen: list[list[str]] = [] - monkeypatch.setattr( - "subprocess.run", - lambda cmd, **kw: seen.append(cmd) - or type("R", (), {"returncode": 0, "stderr": ""})(), - ) - up_mod._pip_install(tmp_path / "pkg") - assert "--break-system-packages" not in seen[0] - - def test_prefers_uv_pip_inside_virtualenv( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - monkeypatch.setenv("VIRTUAL_ENV", str(tmp_path / "venv")) - monkeypatch.setattr(up_mod.shutil, "which", lambda name: f"/usr/bin/{name}") - seen: list[list[str]] = [] - monkeypatch.setattr( - "subprocess.run", - lambda cmd, **kw: seen.append(cmd) - or type("R", (), {"returncode": 0, "stderr": ""})(), - ) - up_mod._pip_install(tmp_path / "pkg") - assert seen[0][:3] == ["uv", "pip", "install"] - assert "--break-system-packages" not in seen[0] + def test_missing_xiaohe(self) -> None: + with patch.object(up_mod.shutil, "which", return_value=None): + result = CliRunner().invoke(up_mod.upgrade_cmd, []) + assert result.exit_code != 0 + assert "xiaohe upgrade" in result.output diff --git a/tests/test_version.py b/tests/test_version.py index 21be535..f4e4590 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -3,16 +3,14 @@ import subprocess from pathlib import Path -from click.testing import CliRunner - from myagents.commands import version as ver_mod class TestDescribeTree: - def test_installed_runtime_tree(self) -> None: + def test_legacy_runtime_tree(self) -> None: tree = Path("/home/u/.xiaohe/runtime/v1.2.3/contrib/myagents") mode, detail = ver_mod.describe_tree(tree) - assert mode == "installed" + assert mode == "legacy-runtime" assert detail == "v1.2.3" def test_development_checkout(self, tmp_path: Path) -> None: @@ -33,16 +31,20 @@ class TestGitDescribe: repo.mkdir() subprocess.run(["git", "init", "-q"], cwd=repo, check=True) subprocess.run( - ["git", "-c", "user.email=t@t", "-c", "user.name=t", - "commit", "-q", "--allow-empty", "-m", "init"], - cwd=repo, check=True, + [ + "git", + "-c", + "user.email=t@t", + "-c", + "user.name=t", + "commit", + "-q", + "--allow-empty", + "-m", + "init", + ], + cwd=repo, + check=True, ) result = ver_mod._git_describe(repo) - assert "branch" in result - - -class TestVersionCommand: - def test_runs_and_reports(self) -> None: - result = CliRunner().invoke(ver_mod.version_cmd) - assert result.exit_code == 0 - assert "myagents" in result.output + assert result != "unknown"