From f3d5414bf926660519139ca8f323df64b2a951c7 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Thu, 16 Jul 2026 18:10:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20xiaohe=20=E5=91=BD=E4=BB=A4=E7=BB=84=20?= =?UTF-8?q?+=20settings.json=20=E6=8B=86=E5=88=86=20+=20workspace=20?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=BC=95=E6=93=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - xiaohe：默认转发 default_agent（settings.json default_agent，兼容 myclaude 拼写）， 子命令 init（首用引导 workspace 路径+骨架+sync+启动器）/ sync（runtime→workspace 哈希三态同步：新增/覆盖/本地改动跳过；git 检出拒绝） - settings.py：~/.xiaohe/agent/settings.json 优先，config.json settings 段兜底 - get_workspace_root：env > settings > project_root/workspace > ~/workspace - build_cli 增 offer_install：缺二进制时交互提示安装（claude/codex 有 install_cmd） - 测试 +11（settings/sync 引擎三态/幂等/git 拒绝） --- myagents/commands/sync_workspace.py | 280 ++++++++++++++++++++++++++++ myagents/entrypoints.py | 44 ++++- myagents/launcher.py | 67 ++++++- myagents/project_root.py | 25 ++- myagents/settings.py | 55 ++++++ pyproject.toml | 1 + tests/test_settings.py | 48 +++++ tests/test_sync_workspace.py | 87 +++++++++ 8 files changed, 593 insertions(+), 14 deletions(-) create mode 100644 myagents/commands/sync_workspace.py create mode 100644 myagents/settings.py create mode 100644 tests/test_settings.py create mode 100644 tests/test_sync_workspace.py diff --git a/myagents/commands/sync_workspace.py b/myagents/commands/sync_workspace.py new file mode 100644 index 0000000..5c85a3e --- /dev/null +++ b/myagents/commands/sync_workspace.py @@ -0,0 +1,280 @@ +"""``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.""" + files: dict[str, str] = {} + for rel in MANAGED_PATHS: + src = runtime / rel + if src.is_file(): + files[rel] = _hash_file(src) + elif src.is_dir(): + for path in sorted(src.rglob("*")): + if path.is_file() and not path.is_symlink(): + 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 + if not link_path.is_symlink() and (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]") + + +def _create_launcher() -> None: + """Double-click launcher that opens a terminal running ``xiaohe``.""" + home = Path.home() + if os.uname().sysname == "Darwin": # noqa: PLR2004 — platform check + desktop = home / "Desktop" + if desktop.is_dir(): + launcher = desktop / "XiaoheAgent.command" + launcher.write_text( + "#!/usr/bin/env bash\n" + "# XiaoheAgent launcher — double-click opens Terminal running xiaohe.\n" + "exec xiaohe\n", + encoding="utf-8", + ) + launcher.chmod(0o755) + console.print(f" [green]launcher:[/green] {launcher}") + else: + apps = home / ".local" / "share" / "applications" + apps.mkdir(parents=True, exist_ok=True) + (apps / "XiaoheAgent.desktop").write_text( + "[Desktop Entry]\n" + "Type=Application\n" + "Name=XiaoheAgent\n" + "Comment=Xiaohe Agent terminal\n" + "Exec=xiaohe\n" + "Terminal=true\n" + "Icon=utilities-terminal\n" + "Categories=Utility;\n", + encoding="utf-8", + ) + console.print(f" [green]launcher:[/green] {apps / 'XiaoheAgent.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]" + ) + + _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/entrypoints.py b/myagents/entrypoints.py index 3f40fb5..51d393b 100644 --- a/myagents/entrypoints.py +++ b/myagents/entrypoints.py @@ -1,4 +1,6 @@ -"""Standalone entrypoints for myclaude, mykimi, mycodex, and myhermes.""" +"""Standalone entrypoints for myclaude, mykimi, mycodex, myhermes, xiaohe.""" + +import re from myagents.launcher import build_cli @@ -7,6 +9,31 @@ kimi_cli = build_cli("kimi", prog_name="mykimi") codex_cli = build_cli("codex", prog_name="mycodex") hermes_cli = build_cli("hermes", prog_name="myhermes") +_KNOWN_BACKENDS = ("claude", "kimi", "codex", "hermes") + + +def default_agent() -> str: + """Default backend for the ``xiaohe`` alias. + + Reads ``default_agent`` from xiaohe settings (accepts both "claude" and + "myclaude" spellings); falls back to claude. + """ + from myagents.settings import get_setting + + agent = str(get_setting("default_agent", "") or "") + agent = re.sub(r"^my", "", agent.strip().lower()) + return agent if agent in _KNOWN_BACKENDS else "claude" + + +def build_xiaohe_cli(): + """The ``xiaohe`` command group: agent forwarding + init/sync subcommands.""" + from myagents.commands.sync_workspace import init_cmd, sync_cmd + + xiaohe_cli = build_cli(default_agent(), prog_name="xiaohe", offer_install=True) + xiaohe_cli.add_command(init_cmd) + xiaohe_cli.add_command(sync_cmd) + return xiaohe_cli + def _progs(): from myagents.cli import cli @@ -17,6 +44,7 @@ def _progs(): ("mykimi", lambda: kimi_cli), ("mycodex", lambda: codex_cli), ("myhermes", lambda: hermes_cli), + ("xiaohe", build_xiaohe_cli), ] @@ -58,3 +86,17 @@ def hermes_main() -> None: ensure_completions_installed(_progs()) hermes_cli() + + +def xiaohe_main() -> None: + """Run ``xiaohe``: default-agent launcher plus init/sync subcommands. + + Same options as the corresponding ``my`` command; when the + underlying CLI is missing, offers to install it interactively. + """ + from myagents.commands.completion_install import ( + ensure_completions_installed, + ) + + ensure_completions_installed(_progs()) + build_xiaohe_cli()() diff --git a/myagents/launcher.py b/myagents/launcher.py index e908e8e..d3b60fd 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -28,6 +28,7 @@ _BACKENDS: dict[str, dict] = { "sessions_root": lambda: Path.home() / ".claude" / "projects", "session_pattern": "*.jsonl", "default_args": ["--dangerously-skip-permissions"], + "install_cmd": ["npm", "install", "-g", "@anthropic-ai/claude-code"], "not_found_msg": ( "[red]claude CLI not found in PATH.[/red] Install Claude Code or set " "[cyan]CLAUDE_BIN[/cyan]." @@ -50,6 +51,7 @@ _BACKENDS: dict[str, dict] = { "sessions_root": lambda: Path.home() / ".codex" / "sessions", "session_pattern": "**/*.jsonl", "default_args": ["--dangerously-bypass-approvals-and-sandbox"], + "install_cmd": ["npm", "install", "-g", "@openai/codex"], "not_found_msg": ( "[red]codex CLI not found in PATH.[/red] Install with " "[cyan]npm install -g @openai/codex[/cyan] or set [cyan]CODEX_BIN[/cyan]." @@ -167,12 +169,57 @@ def _exec_tmux(backend: str, chat_cwd: Path, cmd: list[str]) -> None: raise SystemExit(proc.returncode) +def _offer_install(config: dict) -> str | None: + """Interactively offer to install a missing backend CLI. + + Only backends with a known one-shot installer (``install_cmd``) are + offered; others fall back to the manual ``not_found_msg``. Prompts on + /dev/tty so it also works when stdin is not a terminal. Returns the + resolved binary path after a successful install, else None. + """ + install_cmd = config.get("install_cmd") + if not install_cmd: + return None + stderr_console.print( + f"[yellow]{config['binary']} not found.[/yellow] " + f"Install with [cyan]{shlex.join(install_cmd)}[/cyan] ?" + ) + try: + answer = click.prompt( + "Install now? [y/N]", default="n", show_default=True, err=True + ) + except (click.Abort, EOFError): + return None + if answer.strip().lower() not in ("y", "yes"): + return None + installer = shutil.which(install_cmd[0]) + if not installer: + stderr_console.print( + f"[red]{install_cmd[0]} not found[/red] — cannot auto-install " + f"{config['binary']}. Install it manually." + ) + return None + proc = subprocess.run([installer, *install_cmd[1:]], check=False) + if proc.returncode != 0: + stderr_console.print( + f"[red]Install failed (exit {proc.returncode}).[/red]" + ) + return None + return os.environ.get(config["env_bin"]) or shutil.which(config["binary"]) + + def _launch( - backend: str, chat_cwd: Path, extra: list[str], use_tmux: bool = False + backend: str, + chat_cwd: Path, + extra: list[str], + use_tmux: bool = False, + offer_install: bool = False, ) -> None: """Run backend CLI in chat_cwd, forwarding extra args. Never returns.""" config = _BACKENDS[backend] binary = os.environ.get(config["env_bin"]) or shutil.which(config["binary"]) + if not binary and offer_install: + binary = _offer_install(config) if not binary: stderr_console.print(config["not_found_msg"]) raise SystemExit(127) @@ -460,11 +507,17 @@ class LaunchGroup(click.Group): return run.name, run, [] -def build_cli(backend: str, prog_name: str | None = None) -> click.Group: +def build_cli( + backend: str, + prog_name: str | None = None, + offer_install: bool = False, +) -> click.Group: """Build a click CLI that wraps ``backend`` (claude, kimi, or codex). ``prog_name`` is used in --version output. When omitted it defaults to ``myagents `` (suitable for use as a subcommand of ``myagents``). + With ``offer_install`` a missing backend binary triggers an interactive + install prompt instead of a plain error (used by the ``xiaohe`` alias). """ backend_title = backend.capitalize() if backend == "codex": @@ -520,7 +573,7 @@ def build_cli(backend: str, prog_name: str | None = None) -> click.Group: raise SystemExit(0) if ctx.invoked_subcommand is None: - _launch(backend, chat_cwd, [], use_tmux=tmux) + _launch(backend, chat_cwd, [], use_tmux=tmux, offer_install=offer_install) @cli.command(name="__run__", hidden=True) @click.pass_context @@ -530,7 +583,13 @@ def build_cli(backend: str, prog_name: str | None = None) -> click.Group: assert parent is not None chat_cwd = _resolve_chat_cwd(parent.params.get("cwd")) extra = list(ctx.meta.get("passthrough", [])) - _launch(backend, chat_cwd, extra, use_tmux=bool(parent.params.get("tmux"))) + _launch( + backend, + chat_cwd, + extra, + use_tmux=bool(parent.params.get("tmux")), + offer_install=offer_install, + ) cli.help = ( f"Launch {backend_title} in workspace/.\n\n" diff --git a/myagents/project_root.py b/myagents/project_root.py index aed768e..238310d 100644 --- a/myagents/project_root.py +++ b/myagents/project_root.py @@ -59,23 +59,30 @@ def get_project_root() -> Path: return Path.home() / ".myagents" -def get_workspace_root() -> Path: +def get_workspace_root(create: bool = True) -> Path: """ Root of the myagents workspace directory. - Order: MYAGENTS_WORKSPACE_ROOT > project_root/workspace/ > ~/workspace - Creates the directory if it does not exist. + Order: MYAGENTS_WORKSPACE_ROOT > settings.json workspace_root > + project_root/workspace/ > ~/workspace. Creates the directory when + ``create`` (use ``create=False`` for existence checks). """ + from myagents.settings import get_setting + env_root = os.environ.get("MYAGENTS_WORKSPACE_ROOT") if env_root: root = Path(env_root).expanduser().resolve() else: - project_root = get_project_root() - project_workspace = project_root / "workspace" - if project_workspace.is_dir(): - root = project_workspace + configured = str(get_setting("workspace_root", "") or "").strip() + if configured: + root = Path(configured).expanduser().resolve() else: - root = Path.home() / "workspace" + project_workspace = get_project_root() / "workspace" + if project_workspace.is_dir(): + root = project_workspace + else: + root = Path.home() / "workspace" - root.mkdir(parents=True, exist_ok=True) + if create: + root.mkdir(parents=True, exist_ok=True) return root diff --git a/myagents/settings.py b/myagents/settings.py new file mode 100644 index 0000000..448fffc --- /dev/null +++ b/myagents/settings.py @@ -0,0 +1,55 @@ +"""Xiaohe settings: ~/.xiaohe/agent/settings.json with legacy fallback. + +Layout on disk: +- settings.json — non-secret preferences (default_agent, workspace_root, …); + safe to template, back up, or share. +- config.json — secrets only (keys.*). Historically also carried a + "settings" section; that section is still honored as a + fallback when settings.json lacks a key, but new writes + always go to settings.json. +""" + +import json +from pathlib import Path +from typing import Any + +CONFIG_DIR = Path.home() / ".xiaohe" / "agent" +SETTINGS_PATH = CONFIG_DIR / "settings.json" +LEGACY_CONFIG_PATH = CONFIG_DIR / "config.json" + + +def _read_json(path: Path) -> dict: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def load_settings() -> dict: + """Merged settings: legacy config.json "settings" < settings.json.""" + merged: dict = {} + legacy = _read_json(LEGACY_CONFIG_PATH).get("settings") + if isinstance(legacy, dict): + merged.update(legacy) + merged.update(_read_json(SETTINGS_PATH)) + return merged + + +def save_settings(settings: dict) -> None: + """Persist the full settings dict to settings.json.""" + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + SETTINGS_PATH.write_text( + json.dumps(settings, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + +def get_setting(key: str, default: Any = "") -> Any: + return load_settings().get(key, default) + + +def set_setting(key: str, value: Any) -> None: + settings = load_settings() + settings[key] = value + save_settings(settings) diff --git a/pyproject.toml b/pyproject.toml index 66cba9e..9644098 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ myclaude = "myagents.entrypoints:claude_main" mykimi = "myagents.entrypoints:kimi_main" mycodex = "myagents.entrypoints:codex_main" myhermes = "myagents.entrypoints:hermes_main" +xiaohe = "myagents.entrypoints:xiaohe_main" [dependency-groups] dev = ["pytest>=8.0"] diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..ef7fe7a --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,48 @@ +"""Tests for myagents.settings.""" + +import json +from pathlib import Path + +import pytest + +from myagents import settings as settings_mod + + +@pytest.fixture() +def fake_config_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + config_dir = tmp_path / ".xiaohe" / "agent" + monkeypatch.setattr(settings_mod, "CONFIG_DIR", config_dir) + monkeypatch.setattr(settings_mod, "SETTINGS_PATH", config_dir / "settings.json") + monkeypatch.setattr( + settings_mod, "LEGACY_CONFIG_PATH", config_dir / "config.json" + ) + return config_dir + + +class TestLoadSettings: + def test_empty_when_no_files(self, fake_config_dir: Path) -> None: + assert settings_mod.load_settings() == {} + + def test_settings_json_wins_over_legacy(self, fake_config_dir: Path) -> None: + fake_config_dir.mkdir(parents=True) + (fake_config_dir / "config.json").write_text( + json.dumps({"keys": {}, "settings": {"a": "legacy", "b": "legacy"}}) + ) + (fake_config_dir / "settings.json").write_text(json.dumps({"b": "new"})) + assert settings_mod.load_settings() == {"a": "legacy", "b": "new"} + + def test_corrupt_files_yield_empty(self, fake_config_dir: Path) -> None: + fake_config_dir.mkdir(parents=True) + (fake_config_dir / "settings.json").write_text("{not json") + assert settings_mod.load_settings() == {} + + +class TestSaveAndSet: + def test_set_setting_roundtrip(self, fake_config_dir: Path) -> None: + settings_mod.set_setting("workspace_root", "/tmp/ws") + assert settings_mod.get_setting("workspace_root") == "/tmp/ws" + on_disk = json.loads((fake_config_dir / "settings.json").read_text()) + assert on_disk == {"workspace_root": "/tmp/ws"} + + def test_get_setting_default(self, fake_config_dir: Path) -> None: + assert settings_mod.get_setting("missing", "dflt") == "dflt" diff --git a/tests/test_sync_workspace.py b/tests/test_sync_workspace.py new file mode 100644 index 0000000..7c789aa --- /dev/null +++ b/tests/test_sync_workspace.py @@ -0,0 +1,87 @@ +"""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"]