feat: xiaohe 命令组 + settings.json 拆分 + workspace 同步引擎
- 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 拒绝）
This commit is contained in:
Zhengshou Lai
@@ -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 (<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."""
|
||||
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)
|
||||
+43
-1
@@ -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<agent>`` 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()()
|
||||
|
||||
+63
-4
@@ -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 <backend>`` (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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user