hermes-agent/hermes_cli/agent_import_sync.py

"""``hermes import-agent --sync`` — keep previously imported Claude Code / Codex setups current.

Every successful ``hermes import-agent`` run records its source in ``HERMES_HOME/import-sync.json``
(the sync manifest); ``--sync`` re-imports every registered source whose files changed since the
last run. Change detection is a content digest over exactly the files the importer reads, so an
unchanged source is a cheap no-op and credential files (never read by the importer) can never
trigger a sync. Sync runs prompt-free — it only re-applies sources the user already imported once.
"""

from __future__ import annotations

import hashlib
import json
import logging
import time
from pathlib import Path
from typing import Any, Dict, Iterator, Optional

from utils import atomic_write_text

logger = logging.getLogger(__name__)

SYNC_MANIFEST_NAME = "import-sync.json"
# Credential stores that live inside the source trees; never part of the digest.
_CREDENTIAL_FILENAMES = frozenset({".credentials.json", "auth.json", "credentials.json"})


def sync_manifest_path(target_root: Path) -> Path:
    return Path(target_root) / SYNC_MANIFEST_NAME


def _iter_sync_files(agent: str, source_root: Path) -> Iterator[Path]:
    """Yield the source files whose content determines the sync digest — exactly what
    :class:`~hermes_cli.agent_import.AgentImporter` reads for ``agent``."""
    if agent == "claude-code":
        # ~/.claude.json lives NEXT TO ~/.claude/ (see AgentImporter._run_claude_code)
        paths = [source_root / "CLAUDE.md", source_root / "settings.json",
                 source_root.parent / ".claude.json"]
    else:
        paths = [source_root / "AGENTS.md", source_root / "config.toml"]
        paths.extend(sorted((source_root / "memories").glob("*.md")))
    paths.extend(p for p in sorted((source_root / "skills").rglob("*")) if p.is_file())
    for path in paths:
        if path.name not in _CREDENTIAL_FILENAMES and path.is_file():
            yield path


def compute_source_digest(agent: str, source_root: Path) -> str:
    """Content digest of everything the importer would read from ``source_root``."""
    source_root = Path(source_root)
    digest = hashlib.sha256()
    for path in _iter_sync_files(agent, source_root):
        try:
            rel = str(path.relative_to(source_root))
        except ValueError:
            rel = f"../{path.name}"
        try:
            content_hash = hashlib.sha256(path.read_bytes()).hexdigest()
        except OSError:
            content_hash = "unreadable"
        digest.update(f"{rel}\0{content_hash}\0".encode("utf-8", "replace"))
    return digest.hexdigest()


def skill_tree_digest(skill_dir: Path) -> str:
    """Content digest of an installed skill directory (relative path + bytes of every file)."""
    digest = hashlib.sha256()
    for path in sorted(p for p in Path(skill_dir).rglob("*") if p.is_file()):
        try:
            digest.update(f"{path.relative_to(skill_dir)}\0{hashlib.sha256(path.read_bytes()).hexdigest()}\0".encode("utf-8", "replace"))
        except OSError:
            digest.update(b"unreadable\0")
    return digest.hexdigest()


def managed_skills(entry: Any) -> Dict[str, Optional[str]]:
    """``imported_skills`` as ``{name: digest-at-import}`` (a pre-digest list maps to ``None`` = trusted)."""
    skills = entry.get("imported_skills") if isinstance(entry, dict) else None
    if isinstance(skills, dict):
        return dict(skills)
    return {name: None for name in skills} if isinstance(skills, list) else {}


def load_sync_manifest(target_root: Path) -> Dict[str, Any]:
    """Read the manifest; a missing, unreadable or malformed file yields an empty manifest."""
    path = sync_manifest_path(target_root)
    if not path.exists():
        return {"version": 1, "agents": {}}
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig", errors="replace"))
    except (json.JSONDecodeError, OSError):
        logger.warning("Unreadable %s — starting a fresh sync manifest", path)
        return {"version": 1, "agents": {}}
    if not isinstance(data, dict) or not isinstance(data.get("agents"), dict):
        return {"version": 1, "agents": {}}
    return data


def save_sync_manifest(target_root: Path, manifest: Dict[str, Any]) -> None:
    atomic_write_text(sync_manifest_path(target_root),
                      json.dumps(manifest, indent=2, sort_keys=True) + "\n")


def update_sync_manifest(agent: str, source_root: Path, target_root: Path,
                         overwrite: bool, report: Dict[str, Any], *, refresh_digest: bool = True) -> None:
    """Record/refresh the sync entry for ``agent`` after a real (non-dry-run) import.

    ``imported_skills`` maps every skill this command ever copied for the agent to the digest of
    the copy it wrote: a later sync replaces the destination only while it still matches that
    digest, so a skill the user edited (or created) under the import category is never clobbered.
    ``refresh_digest=False`` keeps the previous source digest so a run with errors is retried.
    """
    manifest = load_sync_manifest(target_root)
    agents = manifest.setdefault("agents", {})
    entry = agents.get(agent)
    if not isinstance(entry, dict):
        entry = {}
    skills = managed_skills(entry)
    for item in report.get("items", []):
        if item.get("kind") == "skill" and item.get("status") == "imported" and item.get("destination"):
            skills[Path(item["destination"]).name] = skill_tree_digest(Path(item["destination"]))
    entry.update({"source": str(source_root), "overwrite": bool(overwrite),
                  "last_import": int(time.time()), "imported_skills": dict(sorted(skills.items()))})
    if refresh_digest or "digest" not in entry:
        entry["digest"] = compute_source_digest(agent, Path(source_root))
    agents[agent] = entry
    save_sync_manifest(target_root, manifest)


def sync_imported_agents(args) -> None:
    """Handle ``hermes import-agent --sync``: re-import every registered source whose digest
    changed. Prompt-free (cron-friendly); ``--dry-run`` previews without touching the manifest."""
    from hermes_cli.agent_import import AgentImporter, print_import_report
    from hermes_cli.setup import print_error, print_header, print_info, print_success
    from hermes_constants import get_hermes_home

    dry_run = bool(getattr(args, "dry_run", False))
    hermes_home = get_hermes_home().resolve()
    agents: Dict[str, Any] = load_sync_manifest(hermes_home).get("agents", {})
    if not agents:
        print()
        print_info("No import sources registered yet.")
        print_info("Run 'hermes import-agent' first — successful imports are "
                   "registered for sync automatically.")
        return

    print()
    print_header("Import Sync")
    synced = unchanged = failed = 0
    for agent_name in sorted(agents):
        entry = agents[agent_name]
        if not isinstance(entry, dict) or not entry.get("source"):
            continue
        source_dir = Path(entry["source"])
        if not source_dir.is_dir():
            print_info(f"{agent_name}: source {source_dir} no longer exists — skipped")
            continue
        if compute_source_digest(agent_name, source_dir) == entry.get("digest"):
            print_info(f"{agent_name}: unchanged since last import")
            unchanged += 1
            continue
        print_info(f"{agent_name}: changes detected in {source_dir}"
                   + (" (dry run)" if dry_run else " — re-importing"))
        overwrite = bool(entry.get("overwrite", False))
        try:
            report = AgentImporter(agent_name, source_dir, hermes_home, execute=not dry_run,
                                   overwrite=overwrite, sync_skills=managed_skills(entry)).run()
        except Exception as exc:  # noqa: BLE001 — keep syncing the other sources
            print_error(f"{agent_name}: sync failed: {exc}")
            logger.debug("import-agent sync error", exc_info=True)
            failed += 1
            continue
        print_import_report(report, dry_run=dry_run)
        had_errors = bool(report.get("summary", {}).get("error"))
        if had_errors:
            failed += 1
        if not dry_run:
            try:
                # Errors keep the old source digest so the next sync retries the failed items.
                update_sync_manifest(agent_name, source_dir, hermes_home, overwrite, report,
                                     refresh_digest=not had_errors)
            except OSError as exc:
                logger.warning("Could not update import sync manifest: %s", exc)
        synced += 1

    print()
    if synced == 0 and failed == 0:
        print_success(f"All {unchanged} source(s) already up to date.")
        return
    parts = [f"{synced} synced"] + [f"{n} {label}" for n, label in
                                    ((unchanged, "unchanged"), (failed, "failed")) if n]
    print_success("Sync complete: " + ", ".join(parts) + ".")