376 lines
14 KiB
Python
376 lines
14 KiB
Python
"""``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)
|