apaam/myagents
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: drop workspace sync launcher path; simplify upgrade/switch

Browse Source 
Align with xiaohe wheel/pip install: remove sync_workspace and desktop
launcher coupling from upgrade/switch flows.
This commit is contained in:
Zhengshou Lai
2026-07-25 12:43:24 +08:00
parent 37961feb21
commit d99c5534cb
11 changed files with 151 additions and 1428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
myagents/commands/info.py
+11 -10
View File
@@ -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 <name>")
console.print(" xiaohe switch provider <name>")
myagents/commands/switch.py
+5 -96
View File
@@ -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 <subcommand> --help' for details.[/dim]")
console.print(
"[dim]Package version: xiaohe upgrade [<version>][/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)
myagents/commands/sync_workspace.py
-375
View File
@@ -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 (<workspace>/.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 = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>CFBundleName</key><string>Xiaohe Agent</string>
<key>CFBundleDisplayName</key><string>Xiaohe Agent</string>
<key>CFBundleIdentifier</key><string>com.xiaohe.agent</string>
<key>CFBundleVersion</key><string>1</string>
<key>CFBundlePackageType</key><string>APPL</string>
<key>CFBundleExecutable</key><string>XiaoheAgent</string>
<key>CFBundleIconFile</key><string>XiaoheAgent</string>
<key>LSMinimumSystemVersion</key><string>11.0</string>
<key>NSHighResolutionCapable</key><true/>
</dict>
</plist>
"""
_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)
myagents/commands/uninstall.py
+7 -10
View File
@@ -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,
myagents/commands/upgrade.py
+25 -265
View File
@@ -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/<version> 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 "<sha256> <name>" 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."
)
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')"
"Install xiaohe-agent, then run: xiaohe upgrade"
)
cmd = [xiaohe, "upgrade"]
if version:
cmd.append(version)
if force:
cmd.append("--force")
raise SystemExit(subprocess.call(cmd))
myagents/commands/version.py
+42 -36
View File
@@ -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:
# <repo>/myagents/__init__.py -> <repo>
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 <repo>/contrib/mytoolkit -e <repo>/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]"
)
myagents/entrypoints.py
+22 -5
View File
@@ -2,6 +2,8 @@
import re
import click
from myagents.launcher import build_cli
claude_cli = build_cli("claude", prog_name="myclaude")
@@ -26,15 +28,31 @@ 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
@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)
@@ -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<agent>`` 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,
tests/test_switch.py
+4 -125
View File
@@ -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
tests/test_sync_workspace.py
-210
View File
@@ -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"
tests/test_upgrade.py
+15 -278
View File
@@ -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"]
)
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
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"]
call.assert_called_once_with(
["/bin/xiaohe", "upgrade", "0.5.1", "--force"]
)
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 "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]
assert "xiaohe upgrade" in result.output
tests/test_version.py
+17 -15
View File
@@ -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"