fix(import-sync): never clobber a locally edited imported skill; keep digest on errors

The docs promised "skills you created or modified under the import category
yourself are never clobbered", but sync replaced any destination whose name was
in `imported_skills`, regardless of what was there now — an imported skill the
user had since edited was silently overwritten on the next source change.

The manifest now stores `imported_skills` as {name: digest-of-the-copy-we-wrote}
and `--sync` refreshes a destination only while it still matches that digest;
a locally modified copy records a `conflict` ("modified locally — not
refreshed") and is skipped. Pre-digest manifests (a plain list) keep the old
trusted behaviour for one more cycle and are upgraded on the next import.

`sync_imported_agents` also refreshed the source digest after a run with
errors, so the failed items were never retried; the previous digest is kept
whenever the report has errors.

hermes_cli/agent_import.py

@@ -15,7 +15,7 @@ import sys
 import time
 import tomllib
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Sequence, Tuple
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
 
 import yaml
 
@@ -236,7 +236,7 @@ class AgentImporter:
 
     def __init__(self, agent: str, source_root: Path, target_root: Path,
                  execute: bool = False, overwrite: bool = False,
-                 sync_skills: Sequence[str] = ()) -> None:
+                 sync_skills: Mapping[str, Optional[str]] | Sequence[str] = ()) -> None:
         if agent not in SUPPORTED_AGENTS:
             raise ValueError(f"Unsupported agent: {agent!r}")
         self.agent = agent
@@ -244,9 +244,11 @@ class AgentImporter:
         self.target_root = Path(target_root)
         self.execute = execute
         self.overwrite = overwrite
-        # Skills a previous import-agent run copied (from the sync manifest): Hermes owns those
-        # destinations, so --sync refreshes them in place; anything else keeps conflict semantics.
-        self.sync_skills = frozenset(sync_skills)
+        # Skills a previous import-agent run copied (from the sync manifest), name → digest of the copy
+        # it wrote (None = pre-digest manifest, trusted). --sync refreshes a destination in place only
+        # while it still matches that digest; a locally edited copy keeps conflict semantics.
+        self.sync_skills: Dict[str, Optional[str]] = (dict(sync_skills) if isinstance(sync_skills, Mapping)
+                                                       else {name: None for name in sync_skills})
         self.items: List[Dict[str, Any]] = []
         self.stripped_secrets: List[str] = []
 
@@ -497,13 +499,18 @@ class AgentImporter:
             self.record("skills", source_root, destination_root, "skipped",
                         "No skills with SKILL.md found")
             return
+        from hermes_cli.agent_import_sync import skill_tree_digest
         for skill_dir in skill_dirs:
             destination = destination_root / skill_dir.name
-            may_replace = self.overwrite or skill_dir.name in self.sync_skills
-            if destination.exists() and not may_replace:
-                self.record("skill", skill_dir, destination, "conflict",
-                            "Destination skill already exists")
-                continue
+            if destination.exists() and not self.overwrite:
+                if skill_dir.name not in self.sync_skills:
+                    self.record("skill", skill_dir, destination, "conflict", "Destination skill already exists")
+                    continue
+                expected = self.sync_skills[skill_dir.name]
+                if expected is not None and skill_tree_digest(destination) != expected:
+                    self.record("skill", skill_dir, destination, "conflict",
+                                "Imported skill was modified locally — not refreshed")
+                    continue
 
             def copy(skill_dir=skill_dir, destination=destination) -> None:
                 destination.parent.mkdir(parents=True, exist_ok=True)

hermes_cli/agent_import_sync.py

@@ -14,7 +14,7 @@ import json
 import logging
 import time
 from pathlib import Path
-from typing import Any, Dict, Iterator, List
+from typing import Any, Dict, Iterator, Optional
 
 from utils import atomic_write_text
 
@@ -62,6 +62,25 @@ def compute_source_digest(agent: str, source_root: Path) -> str:
     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)
@@ -83,26 +102,27 @@ def save_sync_manifest(target_root: Path, manifest: Dict[str, Any]) -> None:
 
 
 def update_sync_manifest(agent: str, source_root: Path, target_root: Path,
-                         overwrite: bool, report: Dict[str, Any]) -> None:
+                         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`` accumulates every skill name this command ever copied for the agent:
-    those destinations are Hermes-owned and may be refreshed in place by a later sync, while
-    skills the user created under the import category keep normal conflict semantics.
+    ``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 = {}
-    previous = entry.get("imported_skills")
-    skills = set(previous) if isinstance(previous, list) else set()
-    skills.update(Path(item["destination"]).name for item in report.get("items", [])
-                  if item.get("kind") == "skill" and item.get("status") == "imported"
-                  and item.get("destination"))
+    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),
-                  "digest": compute_source_digest(agent, Path(source_root)),
-                  "last_import": int(time.time()), "imported_skills": sorted(skills)})
+                  "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)
 
@@ -142,23 +162,23 @@ def sync_imported_agents(args) -> None:
         print_info(f"{agent_name}: changes detected in {source_dir}"
                    + (" (dry run)" if dry_run else " — re-importing"))
         overwrite = bool(entry.get("overwrite", False))
-        sync_skills: List[str] = entry.get("imported_skills") or []
         try:
             report = AgentImporter(agent_name, source_dir, hermes_home, execute=not dry_run,
-                                   overwrite=overwrite,
-                                   sync_skills=sync_skills if isinstance(sync_skills, list) else ()
-                                   ).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)
-        if report.get("summary", {}).get("error"):
+        had_errors = bool(report.get("summary", {}).get("error"))
+        if had_errors:
             failed += 1
         if not dry_run:
             try:
-                update_sync_manifest(agent_name, source_dir, hermes_home, overwrite, report)
+                # 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

tests/hermes_cli/test_agent_import.py

@@ -749,7 +749,7 @@ class TestSyncManifest:
         self._run_command("claude-code", claude_tree)
         entry = load_sync_manifest(hermes_home)["agents"]["claude-code"]
         assert entry["source"] == str(claude_tree.resolve())
-        assert "deploy-helper" in entry["imported_skills"]
+        assert "deploy-helper" in entry["imported_skills"]  # name → digest of the copy we wrote
         # A token refresh in the credential file is invisible to the digest.
         (claude_tree / ".credentials.json").write_text(
             json.dumps({"api_key": "rotated-token"}), encoding="utf-8")
@@ -782,6 +782,14 @@ class TestSyncManifest:
         assert "Deploy v2." in (imports / "deploy-helper" / "SKILL.md").read_text(encoding="utf-8")
         assert (user_skill / "SKILL.md").read_text(encoding="utf-8") == "user content"
 
+        # An imported skill the user then EDITED locally is no longer Hermes-owned: the next sync
+        # records a conflict for it instead of overwriting the edit (the docs promise this).
+        (imports / "deploy-helper" / "SKILL.md").write_text("my local tweaks", encoding="utf-8")
+        (claude_tree / "skills" / "deploy-helper" / "SKILL.md").write_text(
+            "---\nname: deploy-helper\n---\n\nDeploy v3.\n", encoding="utf-8")
+        self._run_command(None, None, sync=True)
+        assert (imports / "deploy-helper" / "SKILL.md").read_text(encoding="utf-8") == "my local tweaks"
+
     def test_sync_dry_run_previews_without_writing(self, claude_tree, hermes_home, capsys):
         from hermes_cli.agent_import_sync import load_sync_manifest