feat(skills): org-skill namespace — token-gated discovery, fail-loud collisions, provenance (M2)

Implements the agreed design (2026-07-23): org skills are FIRST-CLASS (bare
names) with three hard companions.

1. TOKEN-GATED RESOLUTION: _org/<org_id>/ mirrors resolve ONLY while marked
   active. pull_org_skills (which runs only after the token's org_id+org_role
   verified) writes _org/.active_org; discovery (iter_skill_index_files,
   _find_skill_dir, snapshot manifest) prunes every other mirror. Leave the
   org (verified personal token in maybe_pull_org_skills) => marker cleared
   => org skills stop resolving; offline => marker untouched (grace).
   Snapshot manifest includes the marker so org switches invalidate the
   prompt snapshot; _SKILLS_SNAPSHOT_VERSION bumped to 2.

2. FAIL-LOUD COLLISIONS: listing pass unified across snapshot/scan paths;
   a personal/org name clash flags BOTH entries '[name collision — load via
   category path]' — neither side silently wins (personal-wins = silent
   divergence from the org set; org-wins = shadowed personal work).
   skill_view's existing multi-candidate refusal already rejects the
   ambiguous bare name.

3. PROVENANCE: org entries list under an org:<org_id> category with
   '[org-shared: by <author>]' tags; skill_view prepends a load-time header
   (org, author, as-of + read-only/fork-and-propose guidance) INTO the
   content the model consumes, plus an org_provenance result field. Author
   comes from the pull-time .org-provenance.json sidecar (HEAD commit author
   — token-verified at push by the plane's author_mismatch guard, gg #166).

4. READ-ONLY MIRROR: skill_manage patch/edit/delete/write_file refuse org-
   mirror targets with fork-and-propose guidance; org skills are curation-
   exempt (is_curation_eligible False — the org HEAD owns them).

Tests: 11 new (tests/agent/test_org_skill_namespace.py) covering gating,
stale-mirror pruning, org-switch flip, snapshot provenance, listing labels,
both-sides collision flags, read-only guard, curation exemption; 448 green
across skills/prompt/sync suites. Live E2E (real modules, temp HERMES_HOME,
mock plane): merge -> pull -> marker+sidecar -> labeled listing -> exactly-2
collision flags -> load-time header -> edit refused -> marker cleared =>
org skills vanish, personal survive.

agent/prompt_builder.py

@@ -19,13 +19,18 @@ from typing import Optional
 from agent.runtime_cwd import resolve_agent_cwd
 from agent.skill_utils import (
     EXCLUDED_SKILL_DIRS,
+    ORG_ACTIVE_MARKER,
+    ORG_MIRROR_DIR_NAME,
+    ORG_PROVENANCE_FILE,
     SKILL_SUPPORT_DIRS,
     extract_skill_conditions,
     extract_skill_description,
     get_all_skills_dirs,
     get_disabled_skill_names,
     iter_skill_index_files,
+    org_id_of_path,
     parse_frontmatter,
+    read_active_org_id,
     skill_matches_environment,
     skill_matches_platform,
     skill_matches_platform_list,
@@ -1310,7 +1315,9 @@ def drain_truncation_warnings() -> list:
 _SKILLS_PROMPT_CACHE_MAX = 8
 _SKILLS_PROMPT_CACHE: OrderedDict[tuple, str] = OrderedDict()
 _SKILLS_PROMPT_CACHE_LOCK = threading.Lock()
-_SKILLS_SNAPSHOT_VERSION = 1
+# v2: entries gained org provenance fields (org_id/org_author/rel_dir) for M2
+# org-shared skills; older snapshots are discarded and rebuilt.
+_SKILLS_SNAPSHOT_VERSION = 2
 
 
 def _skills_prompt_snapshot_path() -> Path:
@@ -1329,13 +1336,32 @@ def clear_skills_system_prompt_cache(*, clear_snapshot: bool = False) -> None:
 
 
 def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
-    """Build an mtime/size manifest of all SKILL.md and DESCRIPTION.md files."""
+    """Build an mtime/size manifest of all SKILL.md and DESCRIPTION.md files.
+
+    Org mirrors (M2): only the ACTIVE org's mirror participates, and the
+    ``.active_org`` marker itself is included — so switching/leaving an org
+    invalidates the snapshot even when no SKILL.md changed.
+    """
     manifest: dict[str, list[int]] = {}
     skills_dir_str = str(skills_dir)
     base = os.path.join(skills_dir_str, "")
     prefix_len = len(base)
+    active_org = read_active_org_id(skills_dir)
+    org_root = os.path.join(skills_dir_str, ORG_MIRROR_DIR_NAME)
+    marker_path = os.path.join(org_root, ORG_ACTIVE_MARKER)
+    try:
+        st = os.stat(marker_path)
+        manifest[ORG_MIRROR_DIR_NAME + "/" + ORG_ACTIVE_MARKER] = [
+            int(st.st_mtime), int(st.st_size),
+        ]
+    except OSError:
+        pass
     for root, dirs, files in os.walk(skills_dir_str, followlinks=True):
         has_skill_md = "SKILL.md" in files
+        if root == skills_dir_str and ORG_MIRROR_DIR_NAME in dirs and active_org is None:
+            dirs.remove(ORG_MIRROR_DIR_NAME)
+        elif root == org_root:
+            dirs[:] = [d for d in dirs if d == active_org]
         dirs[:] = [
             d
             for d in dirs
@@ -1400,6 +1426,15 @@ def _build_snapshot_entry(
     """Build a serialisable metadata dict for one skill."""
     rel_path = skill_file.relative_to(skills_dir)
     parts = rel_path.parts
+
+    # M2 org mirror: strip the `_org/<org_id>/` prefix so category/name derive
+    # from the path WITHIN the mirror (same shape the org tree was built
+    # from), and record provenance for labeling + fail-loud collisions.
+    org_id: str | None = None
+    if len(parts) >= 3 and parts[0] == ORG_MIRROR_DIR_NAME:
+        org_id = parts[1]
+        parts = parts[2:]
+
     if len(parts) >= 2:
         skill_name = parts[-2]
         category = "/".join(parts[:-2]) if len(parts) > 2 else parts[0]
@@ -1411,7 +1446,7 @@ def _build_snapshot_entry(
     if isinstance(platforms, str):
         platforms = [platforms]
 
-    return {
+    entry = {
         "skill_name": skill_name,
         "category": category,
         "frontmatter_name": str(frontmatter.get("name", skill_name)),
@@ -1419,6 +1454,22 @@ def _build_snapshot_entry(
         "platforms": [str(p).strip() for p in platforms if str(p).strip()],
         "conditions": extract_skill_conditions(frontmatter),
     }
+    if org_id:
+        entry["org_id"] = org_id
+        # Author from the pull-time provenance sidecar (token-verified at
+        # push by the plane's author_mismatch guard). Best-effort.
+        try:
+            import json as _json
+
+            prov_path = (
+                skills_dir / ORG_MIRROR_DIR_NAME / org_id / ORG_PROVENANCE_FILE
+            )
+            prov = _json.loads(prov_path.read_text(encoding="utf-8"))
+            device = str(prov.get("author_device") or "")
+            entry["org_author"] = device or str(prov.get("author_user_id") or "")
+        except Exception:
+            entry["org_author"] = ""
+    return entry
 
 
 # =========================================================================
@@ -1554,6 +1605,10 @@ def build_skills_system_prompt(
 
     skills_by_category: dict[str, list[tuple[str, str]]] = {}
     category_descriptions: dict[str, str] = {}
+    # Unified visible-entry list (both paths) so the org labeling +
+    # fail-loud collision pass below runs identically for snapshot and scan.
+    visible_entries: list[dict] = []
+    skill_entries: list[dict] = []
 
     if snapshot is not None:
         # Fast path: use pre-parsed metadata from disk
@@ -1561,7 +1616,6 @@ def build_skills_system_prompt(
             if not isinstance(entry, dict):
                 continue
             skill_name = entry.get("skill_name") or ""
-            category = entry.get("category") or "general"
             frontmatter_name = entry.get("frontmatter_name") or skill_name
             platforms = entry.get("platforms") or []
             if not skill_matches_platform_list(platforms):
@@ -1574,16 +1628,13 @@ def build_skills_system_prompt(
                 available_toolsets,
             ):
                 continue
-            skills_by_category.setdefault(category, []).append(
-                (frontmatter_name, entry.get("description", ""))
-            )
+            visible_entries.append(entry)
         category_descriptions = {
             str(k): str(v)
             for k, v in (snapshot.get("category_descriptions") or {}).items()
         }
     else:
         # Cold path: full filesystem scan + write snapshot for next time
-        skill_entries: list[dict] = []
         for skill_file in iter_skill_index_files(skills_dir, "SKILL.md"):
             is_compatible, frontmatter, desc = _parse_skill_file(skill_file)
             entry = _build_snapshot_entry(skill_file, skills_dir, frontmatter, desc)
@@ -1599,10 +1650,38 @@ def build_skills_system_prompt(
                 available_toolsets,
             ):
                 continue
-            skills_by_category.setdefault(entry["category"], []).append(
-                (entry["frontmatter_name"], entry["description"])
-            )
+            visible_entries.append(entry)
 
+    # ── M2 org labeling + FAIL-LOUD collisions ─────────────────────────
+    # An org skill lists with an explicit provenance tag. When a personal and
+    # an org skill share a name, NEITHER silently wins: both list qualified
+    # (personal keeps the bare name is the wrong default — silent divergence
+    # from the org set; org winning silently shadows the user's own work) —
+    # so both entries carry a [name collision] flag and skill_view refuses
+    # the ambiguous bare name (its existing multi-candidate guard).
+    name_owners: dict[str, set[str]] = {}
+    for entry in visible_entries:
+        fm = entry.get("frontmatter_name") or entry.get("skill_name") or ""
+        kind = "org" if entry.get("org_id") else "personal"
+        name_owners.setdefault(fm, set()).add(kind)
+    for entry in visible_entries:
+        fm = entry.get("frontmatter_name") or entry.get("skill_name") or ""
+        desc = entry.get("description", "")
+        org_id = entry.get("org_id")
+        collided = len(name_owners.get(fm, set())) > 1
+        if org_id:
+            author = entry.get("org_author") or ""
+            tag = f"[org-shared{': by ' + author if author else ''}]"
+            desc = f"{tag} {desc}".strip()
+            category = f"org:{org_id}"
+        else:
+            category = entry.get("category") or "general"
+        if collided:
+            desc = f"[name collision — also exists {'personally' if org_id else 'in your org'}; load via category path] {desc}".strip()
+        skills_by_category.setdefault(category, []).append((fm, desc))
+
+    if snapshot is None:
+        # (continuation of the cold path below: category descriptions + write)
         # Read category-level DESCRIPTION.md files
         for desc_file in iter_skill_index_files(skills_dir, "DESCRIPTION.md"):
             try:

agent/skill_utils.py

@@ -49,6 +49,52 @@ EXCLUDED_SKILL_DIRS = frozenset(
 # archive workflow preserves a complete old skill package under references/.
 SKILL_SUPPORT_DIRS = frozenset(("references", "templates", "assets", "scripts"))
 
+# ── M2 org-shared skills (hsp-1-contract.md §11) ───────────────────────────
+# Org mirrors live under ~/.hermes/skills/_org/<org_id>/. Resolution is
+# TOKEN-GATED via a marker file the sync client writes after verifying the
+# token (skills_sync_client.pull_org_skills): only the marked org's mirror is
+# scanned. No marker ⇒ no org skills load. The marker is plain data (org_id
+# string) so this module stays import-light; the VERIFICATION lives in the
+# sync client, which is the only writer. Offline grace: the marker persists,
+# so already-pulled org skills keep working without connectivity; a VERIFIED
+# org change (or personal-org token) rewrites/removes it.
+
+ORG_MIRROR_DIR_NAME = "_org"
+ORG_ACTIVE_MARKER = ".active_org"
+ORG_PROVENANCE_FILE = ".org-provenance.json"
+
+
+def read_active_org_id(skills_dir: Path) -> Optional[str]:
+    """The org id whose mirror may resolve, or None (no org skills load)."""
+    try:
+        marker = skills_dir / ORG_MIRROR_DIR_NAME / ORG_ACTIVE_MARKER
+        if not marker.exists():
+            return None
+        val = marker.read_text(encoding="utf-8").strip()
+        return val or None
+    except OSError:
+        return None
+
+
+def is_org_mirror_path(path, skills_dir: Path) -> bool:
+    """True when *path* is inside the org mirror (``_org/``)."""
+    try:
+        rel = Path(path).resolve().relative_to(Path(skills_dir).resolve())
+    except (OSError, ValueError):
+        return False
+    return bool(rel.parts) and rel.parts[0] == ORG_MIRROR_DIR_NAME
+
+
+def org_id_of_path(path, skills_dir: Path) -> Optional[str]:
+    """The ``<org_id>`` segment for a path under ``_org/<org_id>/...``."""
+    try:
+        rel = Path(path).resolve().relative_to(Path(skills_dir).resolve())
+    except (OSError, ValueError):
+        return None
+    if len(rel.parts) >= 2 and rel.parts[0] == ORG_MIRROR_DIR_NAME:
+        return rel.parts[1]
+    return None
+
 
 def is_excluded_skill_path(path) -> bool:
     """True if *path* should be skipped by active skill scanners.
@@ -802,11 +848,24 @@ def iter_skill_index_files(skills_dir: Path, filename: str):
     scripts) can contain arbitrary markdown and even archived package
     ``SKILL.md`` files, but they are progressive-disclosure data loaded through
     ``skill_view(..., file_path=...)`` rather than active skill roots.
+
+    M2 org mirrors (``_org/``): TOKEN-GATED resolution. Only the active org's
+    subdir (per the sync-client-written ``.active_org`` marker) is walked;
+    every other ``_org/<id>/`` (stale mirror from a previous org, or no
+    marker at all) is pruned — leave an org and its skills stop resolving,
+    without any manual cleanup.
     """
     skills_dir_str = str(skills_dir)
+    active_org = read_active_org_id(skills_dir)
+    org_root = os.path.join(skills_dir_str, ORG_MIRROR_DIR_NAME)
     matches: list[str] = []
     for root, dirs, files in os.walk(skills_dir_str, followlinks=True):
         has_skill_md = "SKILL.md" in files
+        if root == skills_dir_str and ORG_MIRROR_DIR_NAME in dirs and active_org is None:
+            dirs.remove(ORG_MIRROR_DIR_NAME)
+        elif root == org_root:
+            # Inside _org/: descend ONLY into the active org's mirror.
+            dirs[:] = [d for d in dirs if d == active_org]
         dirs[:] = [
             d
             for d in dirs

cli.py

@@ -13354,6 +13354,16 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
             )
         except Exception:
             pass
+
+        # HSP skill sync — best-effort periodic pull, piggy-backing on the
+        # curator tick. Inert unless the DEV-PHASE gate is open
+        # (tool_gateway_admin) and a sync base URL is configured; swallows all
+        # errors so it never blocks CLI startup.
+        try:
+            from tools.skills_sync_client import maybe_pull_skills
+            maybe_pull_skills()
+        except Exception:
+            pass
         if self.preloaded_skills and not self._startup_skills_line_shown:
             skills_label = ", ".join(self.preloaded_skills)
             self._console_print(

gateway/run.py

@@ -22830,6 +22830,15 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop
             except Exception as e:
                 logger.debug("Curator tick error: %s", e)
 
+            # HSP skill sync — best-effort periodic pull on the same cadence.
+            # Inert unless the DEV-PHASE gate is open (tool_gateway_admin) and
+            # a sync base URL is configured; never raises.
+            try:
+                from tools.skills_sync_client import maybe_pull_skills
+                maybe_pull_skills()
+            except Exception as e:
+                logger.debug("Sync pull tick error: %s", e)
+
         stop_event.wait(timeout=interval)
     logger.info("Gateway housekeeping stopped")
 

hermes_cli/main.py

@@ -402,6 +402,7 @@ from typing import Optional
 
 from hermes_cli.subcommands._shared import add_accept_hooks_flag as _add_accept_hooks_flag
 from hermes_cli.subcommands.cron import build_cron_parser
+from hermes_cli.subcommands.sync import build_sync_parser
 from hermes_cli.subcommands.gateway import build_gateway_parser
 from hermes_cli.subcommands.profile import build_profile_parser
 from hermes_cli.subcommands.model import build_model_parser
@@ -4438,6 +4439,131 @@ def cmd_cron(args):
     cron_command(args)
 
 
+def cmd_sync(args):
+    """HSP/1 personal skill sync management (status/pull/push/now/enable/disable)."""
+    import json as _json
+
+    sub = getattr(args, "sync_command", None)
+
+    if sub in {None, ""}:
+        print(
+            "usage: hermes sync <status|pull|push|now|enable|disable|device>\n"
+            "\n"
+            "  status            Show sync gate, opt-in, and head state\n"
+            "  pull              Pull the owner's HEAD, materialize opted-in skills\n"
+            "  push              Push opted-in skills to the owner's HEAD\n"
+            "  now               Reconcile now: pull then push\n"
+            "  enable <skill>    Opt a skill into sync (M1-D opt-in)\n"
+            "  disable <skill>   Opt a skill out of sync\n"
+            "  device [--name N] Show or set this device's sync label",
+            file=sys.stderr,
+        )
+        return 1
+
+    if sub == "device":
+        from tools import skills_sync_client as ssc
+
+        name = getattr(args, "device_name", None)
+        if name is not None:
+            try:
+                stored = ssc.set_device_name(name)
+            except ValueError as e:
+                print(f"error: {e}", file=sys.stderr)
+                return 1
+            print(f"device label set to '{stored}'.")
+            print(
+                "New commits from this device will use this label; existing "
+                "commits keep their previous one.",
+                file=sys.stderr,
+            )
+            return 0
+        # No --name: print the current (creating a default on first use).
+        print(ssc.stable_device_id())
+        return 0
+
+    if sub in {"enable", "disable"}:
+        from tools.skill_usage import set_sync, is_curation_eligible
+
+        skill = args.skill
+        if not is_curation_eligible(skill):
+            print(
+                f"'{skill}' is not sync-eligible (bundled, hub-installed, "
+                f"external, or not found). Only agent-created / user-authored "
+                f"skills under ~/.hermes/skills/ can sync.",
+                file=sys.stderr,
+            )
+            return 1
+        set_sync(skill, sub == "enable")
+        print(f"sync {'enabled' if sub == 'enable' else 'disabled'} for '{skill}'.")
+        return 0
+
+    from tools import skills_sync_client as ssc
+
+    if sub == "status":
+        status = ssc.sync_status()
+        print(_json.dumps(status, indent=2, ensure_ascii=False))
+        if not status.get("logged_in"):
+            print("\nNot logged into Nous Portal — sync is inert.", file=sys.stderr)
+        elif not status.get("dev_gate_ok"):
+            print(
+                "\nDEV-PHASE gate closed: your token lacks 'tool_gateway_admin'. "
+                "Sync is inert during the dev rollout.",
+                file=sys.stderr,
+            )
+        elif not status.get("feature_enabled"):
+            print(
+                "\nSync feature is off for this instance (set HERMES_SYNC_ENABLED=1 "
+                "or config.yaml sync.enabled: true). Sync is inert.",
+                file=sys.stderr,
+            )
+        elif not status.get("base_url"):
+            print(
+                "\nNo sync base URL configured (config.yaml sync.base_url or "
+                "HERMES_SYNC_BASE_URL). Sync is inert.",
+                file=sys.stderr,
+            )
+        return 0
+
+    # pull / push / now — enforce the gate up front with a clear message.
+    try:
+        identity = ssc.resolve_identity()
+    except ssc.SyncInertError as e:
+        print(f"sync inert: {e}", file=sys.stderr)
+        return 1
+    if not identity.get("dev_gate_ok"):
+        print(
+            "sync inert: DEV-PHASE gate closed (token lacks 'tool_gateway_admin').",
+            file=sys.stderr,
+        )
+        return 1
+    if not ssc.resolve_sync_base_url():
+        print(
+            "sync inert: no sync base URL configured (config.yaml sync.base_url "
+            "or HERMES_SYNC_BASE_URL).",
+            file=sys.stderr,
+        )
+        return 1
+
+    try:
+        if sub == "pull":
+            result = ssc.pull_skills(identity=identity)
+        elif sub == "push":
+            result = ssc.push_skills(identity=identity, message="hermes sync push")
+        elif sub == "now":
+            pull_res = ssc.pull_skills(identity=identity)
+            push_res = ssc.push_skills(identity=identity, message="hermes sync now")
+            result = {"pull": pull_res, "push": push_res}
+        else:
+            print(f"Unknown sync subcommand: {sub}", file=sys.stderr)
+            return 1
+    except ssc.HSPError as e:
+        print(f"sync failed: {e}", file=sys.stderr)
+        return 1
+
+    print(_json.dumps(result, indent=2, ensure_ascii=False))
+    return 0
+
+
 def cmd_webhook(args):
     """Webhook subscription management."""
     from hermes_cli.webhook import webhook_command
@@ -13313,7 +13439,7 @@ _BUILTIN_SUBCOMMANDS = frozenset(
         "project", "proxy",
         "prompt-size",
         "send", "sessions", "setup",
-        "skin", "skills", "slack", "status", "tools", "uninstall", "update",
+        "skin", "skills", "slack", "status", "sync", "tools", "uninstall", "update",
         "version", "webhook", "whatsapp", "whatsapp-cloud", "chat", "secrets", "security",
         # Help-ish invocations — plugin commands not being listed in
         # top-level --help is an acceptable trade-off for skipping an
@@ -13769,6 +13895,34 @@ def cmd_skills(args):
         from hermes_cli.skills_config import skills_command as skills_config_command
 
         skills_config_command(args)
+    elif getattr(args, "skills_action", None) == "propose":
+        # M2 org-shared skills (hsp-1-contract.md §11.5): propose a local
+        # skill to the org canonical set. 202 => pending review (NEVER shown
+        # as live); direct merge for admins. Personal orgs have no org
+        # workflow — say so plainly instead of a raw 403.
+        from tools import skills_sync_client as ssc
+
+        name = args.name
+        try:
+            result = ssc.propose_skill(name, message=args.message)
+        except ssc.SyncInertError as e:
+            print(f"org sync unavailable: {e}", file=sys.stderr)
+            return 1
+        except ssc.HSPError as e:
+            print(f"propose failed: {e}", file=sys.stderr)
+            return 1
+        if result.get("proposal_pending"):
+            print(
+                f"proposed '{name}' — pending admin review "
+                f"(proposal #{result.get('proposal_id')}). Not live for the "
+                f"org until approved."
+            )
+        else:
+            print(
+                f"merged '{name}' into the org set "
+                f"(head {str(result.get('head', ''))[:19]}…)."
+            )
+        return 0
     else:
         from hermes_cli.skills_hub import skills_command
 
@@ -14066,6 +14220,7 @@ def main():
     # cron command  (parser built in hermes_cli/subcommands/cron.py)
     # =========================================================================
     build_cron_parser(subparsers, cmd_cron=cmd_cron)
+    build_sync_parser(subparsers, cmd_sync=cmd_sync)
 
     # =========================================================================
     # webhook command  (parser built in hermes_cli/subcommands/webhook.py)

hermes_cli/subcommands/skills.py

@@ -312,4 +312,27 @@ def build_skills_parser(subparsers, *, cmd_skills: Callable) -> None:
         "config",
         help="Interactive skill configuration — enable/disable individual skills",
     )
+
+    # M2 org-shared skills (hsp-1-contract.md §11.5/§11.11): propose a local
+    # skill's content to the org canonical set. MEMBER → 202 proposal
+    # (pending admin review); ADMIN/OWNER → direct merge. Only meaningful for
+    # multi-member orgs — personal orgs have no org workflow (the command
+    # reports that instead of failing opaquely).
+    skills_propose = skills_subparsers.add_parser(
+        "propose",
+        help="Propose a skill to your org's shared skill set (M2)",
+        description=(
+            "Snapshot the local skill and submit it to the org canonical set. "
+            "An org admin's push merges directly; a member's push becomes a "
+            "proposal reviewed in the org console. Personal orgs keep simple "
+            "personal sync and have no proposal workflow."
+        ),
+    )
+    skills_propose.add_argument("name", help="Skill name to propose")
+    skills_propose.add_argument(
+        "-m",
+        "--message",
+        default=None,
+        help="Optional proposal message (defaults to 'propose <name>')",
+    )
     skills_parser.set_defaults(func=cmd_skills)

hermes_cli/subcommands/sync.py

@@ -0,0 +1,57 @@
+"""``hermes sync`` subcommand parser (HSP/1 personal skill sync).
+
+Cloned from ``hermes_cli/subcommands/cron.py`` — same injected-handler shape
+(``func=cmd_sync``) so this module does not import ``main`` (cycle avoidance).
+
+Commands:
+  hermes sync status   -- show gate/opt-in/head state
+  hermes sync pull      -- pull the owner's HEAD, materialize opted-in skills
+  hermes sync push      -- push opted-in skills to the owner's HEAD
+  hermes sync now       -- pull then push (full reconcile)
+  hermes sync enable <skill>   -- opt a skill into sync (M1-D)
+  hermes sync disable <skill>  -- opt a skill out of sync
+  hermes sync device [--name]  -- show or set this device's sync label
+
+Sync is INERT unless the resolved Nous token carries the DEV-PHASE gate claim
+(tool_gateway_admin) AND a sync base URL is configured. The commands report
+that state rather than failing opaquely.
+"""
+
+from __future__ import annotations
+
+from typing import Callable
+
+
+def build_sync_parser(subparsers, *, cmd_sync: Callable) -> None:
+    """Attach the ``sync`` subcommand (and its sub-actions) to ``subparsers``."""
+    sync_parser = subparsers.add_parser(
+        "sync",
+        help="Personal skill sync (HSP/1)",
+        description="Sync agent-created and user-authored skills across devices.",
+    )
+    sync_sub = sync_parser.add_subparsers(dest="sync_command")
+
+    sync_sub.add_parser("status", help="Show sync gate, opt-in, and head state")
+    sync_sub.add_parser("pull", help="Pull the owner's HEAD and materialize opted-in skills")
+    sync_sub.add_parser("push", help="Push opted-in skills to the owner's HEAD")
+    sync_sub.add_parser("now", help="Reconcile now: pull then push")
+
+    enable = sync_sub.add_parser("enable", help="Opt a skill into sync")
+    enable.add_argument("skill", help="Skill name (frontmatter name / directory name)")
+
+    disable = sync_sub.add_parser("disable", help="Opt a skill out of sync")
+    disable.add_argument("skill", help="Skill name (frontmatter name / directory name)")
+
+    device = sync_sub.add_parser(
+        "device",
+        help="Show or set this device's sync label (shown in the sync console)",
+    )
+    device.add_argument(
+        "--name",
+        dest="device_name",
+        default=None,
+        help="Set a human-friendly label for this device (e.g. \"Ben's Laptop\"). "
+        "Omit to print the current label.",
+    )
+
+    sync_parser.set_defaults(func=cmd_sync)

tests/agent/test_org_skill_namespace.py

@@ -0,0 +1,176 @@
+"""M2 org-skill namespace: token-gated resolution, provenance, collisions.
+
+Covers the design agreed 2026-07-23 (bare-name first-class org skills):
+ 1. TOKEN-GATED discovery — only the `.active_org`-marked mirror resolves;
+    stale mirrors and marker-less trees never load.
+ 2. Fail-loud collisions — a personal/org name clash lists BOTH sides flagged;
+    skill_view's existing multi-candidate guard refuses the bare name.
+ 3. Load-time provenance header — org skill content announces org + author.
+ 4. Org mirrors are read-only (skill_manage guards) and curation-exempt.
+"""
+
+import json
+
+import pytest
+
+from agent import skill_utils as sku
+from agent.prompt_builder import _build_snapshot_entry
+
+
+def _mk_skill(root, rel, name=None, body="# body\n"):
+    d = root
+    for part in rel.split("/"):
+        d = d / part
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "SKILL.md").write_text(
+        f"---\nname: {name or rel.split('/')[-1]}\ndescription: d\n---\n{body}",
+        encoding="utf-8",
+    )
+    return d
+
+
+def _mark_active(skills, org_id):
+    org_root = skills / sku.ORG_MIRROR_DIR_NAME
+    org_root.mkdir(parents=True, exist_ok=True)
+    (org_root / sku.ORG_ACTIVE_MARKER).write_text(org_id, encoding="utf-8")
+
+
+class TestTokenGatedDiscovery:
+    def test_no_marker_no_org_skills(self, tmp_path):
+        skills = tmp_path / "skills"
+        _mk_skill(skills, "personal-a")
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        found = [p.parent.name for p in sku.iter_skill_index_files(skills, "SKILL.md")]
+        assert "personal-a" in found
+        assert "shared-x" not in found  # unmarked mirror never resolves
+
+    def test_marker_gates_to_active_org_only(self, tmp_path):
+        skills = tmp_path / "skills"
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-OLD/stale-y", name="stale-y")
+        _mark_active(skills, "org-1")
+        found = [p.parent.name for p in sku.iter_skill_index_files(skills, "SKILL.md")]
+        assert "shared-x" in found
+        assert "stale-y" not in found  # stale mirror pruned at resolution
+
+    def test_switching_org_flips_resolution(self, tmp_path):
+        skills = tmp_path / "skills"
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-2/other-z", name="other-z")
+        _mark_active(skills, "org-2")
+        found = [p.parent.name for p in sku.iter_skill_index_files(skills, "SKILL.md")]
+        assert found and "other-z" in found and "shared-x" not in found
+
+    def test_helpers(self, tmp_path):
+        skills = tmp_path / "skills"
+        d = _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-9/cat/sk", name="sk")
+        assert sku.is_org_mirror_path(d, skills) is True
+        assert sku.org_id_of_path(d, skills) == "org-9"
+        p = _mk_skill(skills, "plain")
+        assert sku.is_org_mirror_path(p, skills) is False
+        assert sku.read_active_org_id(skills) is None
+        _mark_active(skills, "org-9")
+        assert sku.read_active_org_id(skills) == "org-9"
+
+
+class TestSnapshotEntryProvenance:
+    def test_org_entry_strips_prefix_and_carries_provenance(self, tmp_path):
+        skills = tmp_path / "skills"
+        d = _mk_skill(
+            skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/devops/beta", name="beta"
+        )
+        (skills / sku.ORG_MIRROR_DIR_NAME / "org-1" / sku.ORG_PROVENANCE_FILE).write_text(
+            json.dumps(
+                {"author_device": "bens-macbook-a1b2c3", "author_user_id": "u1"}
+            ),
+            encoding="utf-8",
+        )
+        entry = _build_snapshot_entry(d / "SKILL.md", skills, {"name": "beta"}, "d")
+        assert entry["org_id"] == "org-1"
+        assert entry["org_author"] == "bens-macbook-a1b2c3"
+        # Category derives from the path WITHIN the mirror, not _org/org-1/...
+        assert entry["category"] == "devops"
+        assert entry["skill_name"] == "beta"
+
+    def test_personal_entry_unchanged(self, tmp_path):
+        skills = tmp_path / "skills"
+        d = _mk_skill(skills, "devops/beta", name="beta")
+        entry = _build_snapshot_entry(d / "SKILL.md", skills, {"name": "beta"}, "d")
+        assert "org_id" not in entry
+        assert entry["category"] == "devops"
+
+
+class TestListingCollisionsAndLabels:
+    def _render(self, tmp_path, monkeypatch):
+        from agent import prompt_builder as pb
+
+        skills = tmp_path / "skills"
+        skills.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setattr(pb, "get_skills_dir", lambda: skills, raising=True)
+        monkeypatch.setattr(
+            pb, "get_all_skills_dirs", lambda: [skills], raising=True
+        )
+        monkeypatch.setattr(pb, "get_disabled_skill_names", lambda *a, **k: set())
+        monkeypatch.setattr(
+            pb, "_skills_prompt_snapshot_path", lambda: tmp_path / "snap.json"
+        )
+        pb.clear_skills_system_prompt_cache()
+        return skills, pb
+
+    def test_org_skill_listed_with_provenance_tag(self, tmp_path, monkeypatch):
+        skills, pb = self._render(tmp_path, monkeypatch)
+        _mk_skill(skills, "personal-a")
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        (skills / sku.ORG_MIRROR_DIR_NAME / "org-1" / sku.ORG_PROVENANCE_FILE).write_text(
+            json.dumps({"author_device": "bens-macbook"}), encoding="utf-8"
+        )
+        _mark_active(skills, "org-1")
+        out = pb.build_skills_system_prompt()
+        assert "org:org-1" in out
+        assert "[org-shared: by bens-macbook]" in out
+        assert "personal-a" in out
+
+    def test_collision_flags_both_sides(self, tmp_path, monkeypatch):
+        skills, pb = self._render(tmp_path, monkeypatch)
+        _mk_skill(skills, "k8s-debug", body="personal version\n")
+        _mk_skill(
+            skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/k8s-debug", name="k8s-debug"
+        )
+        _mark_active(skills, "org-1")
+        out = pb.build_skills_system_prompt()
+        # BOTH entries flagged — neither silently wins.
+        assert out.count("[name collision") == 2
+
+    def test_no_collision_flag_when_unique(self, tmp_path, monkeypatch):
+        skills, pb = self._render(tmp_path, monkeypatch)
+        _mk_skill(skills, "personal-a")
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        _mark_active(skills, "org-1")
+        out = pb.build_skills_system_prompt()
+        assert "[name collision" not in out
+
+
+class TestOrgMirrorReadOnly:
+    def test_skill_manage_patch_refuses_org_mirror(self, tmp_path, monkeypatch):
+        from tools import skill_manager_tool as smt
+
+        skills = tmp_path / "skills"
+        _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        _mark_active(skills, "org-1")
+        monkeypatch.setattr(smt, "_skills_dir", lambda: skills)
+        from agent import skill_utils as _sku
+        monkeypatch.setattr(
+            _sku, "get_all_skills_dirs", lambda: [skills], raising=True
+        )
+        result = smt._patch_skill("shared-x", "body", "hacked")
+        assert result["success"] is False
+        assert "ORG-SHARED" in result["error"]
+        assert "propose" in result["error"]
+
+    def test_curation_exempt(self, tmp_path, monkeypatch):
+        from tools import skill_usage as su
+
+        skills = tmp_path / "skills"
+        d = _mk_skill(skills, f"{sku.ORG_MIRROR_DIR_NAME}/org-1/shared-x", name="shared-x")
+        monkeypatch.setattr(su, "_skills_dir", lambda: skills)
+        assert su.is_curation_eligible("shared-x", d) is False

tests/tools/test_skills_sync_client.py

@@ -0,0 +1,951 @@
+"""Tests for tools/skills_sync_client.py — the HSP/1 sync client.
+
+Covers, against the frozen contract (~/src/specs/collective-wisdom/
+hsp-1-contract.md):
+  * content addressing (full 64-hex) + canonical JSON (§2.1, §2.5)
+  * the DEV-PHASE gate (tool_gateway_admin) making sync inert
+  * the M1-D opt-in default (nothing syncs without the sync flag)
+  * object building (blob/tree/commit, exec mode, size limit)
+  * push (upload + CAS), pull (materialize), and the three-way merge / 409
+    conflict paths — all against an in-process mock HSP server.
+
+The mock server implements the contract §3/§4 endpoint shapes with an
+in-memory object store + ref table. No live server, no network.
+"""
+
+import hashlib
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+import pytest
+
+import tools.skills_sync_client as ssc
+
+
+# ---------------------------------------------------------------------------
+# In-process mock HSP/1 server (contract §3-§4)
+# ---------------------------------------------------------------------------
+
+class _MockState:
+    def __init__(self):
+        self.objects = {}   # hash -> (kind, bytes)
+        self.refs = {}      # name -> commit hash
+        self.hsp_version = "1"
+        self.max_object_bytes = 26214400
+        self.force_conflict_once = False  # inject a 409 on the next CAS
+        # M2 org behavior (contract §11): advertise the "org" feature and,
+        # when org_role_admin is False, convert org-HEAD CAS to 202 proposals.
+        self.org_feature = True
+        self.org_role_admin = True
+        self.proposals = []  # [{n, to, base}]
+
+
+def _make_handler(state: _MockState):
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, format, *args):  # silence
+            pass
+
+        def _json(self, code, obj, extra_headers=None):
+            body = json.dumps(obj).encode("utf-8")
+            self.send_response(code)
+            self.send_header("Content-Type", "application/json")
+            for k, v in (extra_headers or {}).items():
+                self.send_header(k, v)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self):
+            path = self.path.split("?", 1)[0]
+            query = ""
+            if "?" in self.path:
+                query = self.path.split("?", 1)[1]
+
+            if path == "/v1/sync/capabilities":
+                features = ["personal"] + (["org"] if state.org_feature else [])
+                return self._json(200, {
+                    "hsp_version": state.hsp_version,
+                    "features": features,
+                    "max_object_bytes": state.max_object_bytes,
+                    "hash_alg": "sha256",
+                    "auth": "bearer",
+                })
+
+            if path == "/v1/sync/refs":
+                prefix = ""
+                for part in query.split("&"):
+                    if part.startswith("prefix="):
+                        from urllib.parse import unquote
+                        prefix = unquote(part[len("prefix="):])
+                refs = [
+                    {"name": n, "hash": h}
+                    for n, h in state.refs.items()
+                    if n.startswith(prefix)
+                ]
+                return self._json(200, {"refs": refs})
+
+            if path.startswith("/v1/sync/objects/"):
+                obj_hash = path[len("/v1/sync/objects/"):]
+                if obj_hash not in state.objects:
+                    return self._json(404, {"error": "not_found"})
+                kind, data = state.objects[obj_hash]
+                if kind == ssc.KIND_BLOB:
+                    self.send_response(200)
+                    self.send_header("Content-Type", "application/octet-stream")
+                    self.send_header("X-HSP-Object-Type", "blob")
+                    self.send_header("Content-Length", str(len(data)))
+                    self.end_headers()
+                    self.wfile.write(data)
+                    return
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("X-HSP-Object-Type", kind)
+                self.send_header("Content-Length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+                return
+
+            self._json(404, {"error": "unknown"})
+
+        def do_POST(self):
+            length = int(self.headers.get("Content-Length", 0))
+            raw = self.rfile.read(length) if length else b""
+            path = self.path.split("?", 1)[0]  # e.g. /v1/sync/objects?scope=org
+
+            if path == "/v1/sync/objects":
+                return self._handle_put_objects(raw)
+
+            if path.startswith("/v1/sync/refs/"):
+                return self._handle_cas(raw)
+
+            self._json(404, {"error": "unknown"})
+
+        def _handle_put_objects(self, raw):
+            # multipart/form-data: parse parts (field=hash, filename=type,
+            # body=raw bytes). The server recomputes each hash and 422s on
+            # mismatch (contract §4.2).
+            ctype = self.headers.get("Content-Type", "")
+            if "multipart/form-data" not in ctype:
+                return self._json(400, {"error": "expected multipart"})
+            boundary = ctype.split("boundary=", 1)[1].encode("ascii")
+            accepted, already = [], []
+            parts = raw.split(b"--" + boundary)
+            for part in parts:
+                # Only trim the delimiter framing: a leading CRLF and a
+                # trailing CRLF. Do NOT strip() the whole part -- that would
+                # also eat legitimate trailing newlines from the object bytes.
+                if part.startswith(b"\r\n"):
+                    part = part[2:]
+                if part.endswith(b"\r\n"):
+                    part = part[:-2]
+                if not part or part == b"--":
+                    continue
+                if b"\r\n\r\n" not in part:
+                    continue
+                headers_blob, body = part.split(b"\r\n\r\n", 1)
+                hdr_text = headers_blob.decode("utf-8", "replace")
+                claimed_hash = None
+                kind = None
+                for line in hdr_text.split("\r\n"):
+                    if line.lower().startswith("content-disposition"):
+                        for token in line.split(";"):
+                            token = token.strip()
+                            if token.startswith('name="'):
+                                claimed_hash = token[len('name="'):-1]
+                            elif token.startswith('filename="'):
+                                kind = token[len('filename="'):-1]
+                if claimed_hash is None:
+                    continue
+                real = "sha256:" + hashlib.sha256(body).hexdigest()
+                if real != claimed_hash:
+                    return self._json(422, {
+                        "error": "hash_mismatch", "claimed": claimed_hash,
+                    })
+                if claimed_hash in state.objects:
+                    already.append(claimed_hash)
+                else:
+                    state.objects[claimed_hash] = (kind, body)
+                    accepted.append(claimed_hash)
+            return self._json(200, {"accepted": accepted, "already_present": already})
+
+        def _handle_cas(self, raw):
+            from urllib.parse import unquote
+            name = unquote(self.path[len("/v1/sync/refs/"):])
+            body = json.loads(raw.decode("utf-8")) if raw else {}
+            frm = body.get("from")
+            to = body.get("to")
+            # M2 (contract §11.5): a non-admin member's CAS on an org HEAD is
+            # accept-always converted to a proposal → 202.
+            if name.startswith("refs/org/") and not state.org_role_admin:
+                n = len(state.proposals) + 1
+                state.proposals.append({"n": n, "to": to, "base": frm})
+                org = name.split("/")[2]
+                prop_ref = f"refs/org/{org}/proposals/{n}"
+                state.refs[prop_ref] = to
+                return self._json(202, {"proposal_id": n, "ref": prop_ref})
+            if state.force_conflict_once:
+                state.force_conflict_once = False
+                return self._json(409, {"actual": state.refs.get(name, "")})
+            current = state.refs.get(name)
+            if current != frm:
+                return self._json(409, {"actual": current or ""})
+            state.refs[name] = to
+            return self._json(200, {"ref": name, "hash": to})
+
+    return Handler
+
+
+@pytest.fixture
+def mock_server():
+    state = _MockState()
+    server = HTTPServer(("127.0.0.1", 0), _make_handler(state))
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    base = f"http://127.0.0.1:{server.server_address[1]}"
+    try:
+        yield base, state
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _write_skill(skills_dir: Path, name: str, body: str = "# skill\n", *, category=None):
+    """Create a minimal skill dir under skills_dir; return its path."""
+    parent = skills_dir / category if category else skills_dir
+    d = parent / name
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "SKILL.md").write_text(
+        f"---\nname: {name}\ndescription: test\n---\n{body}", encoding="utf-8"
+    )
+    return d
+
+
+def _jwt(claims: dict) -> str:
+    import jwt as _pyjwt
+    return _pyjwt.encode(claims, "x" * 32, algorithm="HS256")
+
+
+# ---------------------------------------------------------------------------
+# Content addressing & canonicalization (contract §2.1, §2.5, OI-5)
+# ---------------------------------------------------------------------------
+
+class TestAddressing:
+    def test_full_64_hex_address(self):
+        addr = ssc.hsp_address(b"")
+        # sha256 of empty is the well-known e3b0... digest, full 64 hex.
+        assert addr == (
+            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+        )
+        assert len(addr.split(":", 1)[1]) == 64
+
+    def test_address_differs_from_local_truncated_namespace(self):
+        # OI-5: HSP full-64-hex must NOT equal the local truncated 16-hex form.
+        data = b"hello world"
+        full = ssc.hsp_address(data)
+        truncated = "sha256:" + hashlib.sha256(data).hexdigest()[:16]
+        assert full != truncated
+        assert len(full.split(":")[1]) == 64
+        assert len(truncated.split(":")[1]) == 16
+
+    def test_canonical_json_sorted_no_whitespace(self):
+        out = ssc.canonical_json_bytes({"b": 1, "a": 2})
+        assert out == b'{"a":2,"b":1}'
+        assert b" " not in out
+        assert not out.endswith(b"\n")
+
+    def test_canonical_json_stable(self):
+        obj = {"type": "tree", "entries": [{"name": "x", "hash": "sha256:aa"}]}
+        assert ssc.canonical_json_bytes(obj) == ssc.canonical_json_bytes(dict(obj))
+
+
+# ---------------------------------------------------------------------------
+# DEV-PHASE gate (tool_gateway_admin) + M1-D opt-in
+# ---------------------------------------------------------------------------
+
+class TestDevGate:
+    def test_gate_open_with_claim(self, monkeypatch):
+        token = _jwt({"sub": "user1", "tool_gateway_admin": True})
+        monkeypatch.setattr(
+            ssc, "resolve_nous_runtime_credentials",
+            lambda **kw: {"api_key": token, "base_url": "https://x"}, raising=False,
+        )
+        # patch the lazily-imported symbol used inside resolve_identity
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token, "base_url": "https://x"})
+        ident = ssc.resolve_identity()
+        assert ident["dev_gate_ok"] is True
+        assert ident["owner"] == "user1"
+
+    def test_gate_closed_without_claim(self, monkeypatch):
+        token = _jwt({"sub": "user1"})  # no tool_gateway_admin
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token, "base_url": "https://x"})
+        ident = ssc.resolve_identity()
+        assert ident["dev_gate_ok"] is False
+
+    def test_gate_closed_when_claim_false(self, monkeypatch):
+        token = _jwt({"sub": "u", "tool_gateway_admin": False})
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token, "base_url": "https://x"})
+        assert ssc.dev_gate_open() is False
+
+    def test_maybe_push_inert_when_gate_closed(self, monkeypatch):
+        token = _jwt({"sub": "u"})
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token})
+        monkeypatch.setattr(ssc, "resolve_sync_base_url", lambda: "http://x")
+        # gate closed -> None (inert), never attempts a push
+        assert ssc.maybe_push_skills() is None
+
+    def test_maybe_pull_inert_when_not_logged_in(self, monkeypatch):
+        import hermes_cli.auth as auth_mod
+
+        def _raise(**kw):
+            raise RuntimeError("not logged in")
+
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials", _raise)
+        assert ssc.maybe_pull_skills() is None
+
+
+# ---------------------------------------------------------------------------
+# Object building (contract §2.2-§2.4)
+# ---------------------------------------------------------------------------
+
+class TestObjectBuilding:
+    def test_build_tree_blob_and_exec(self, tmp_path):
+        d = tmp_path / "skill"
+        d.mkdir()
+        (d / "SKILL.md").write_text("hello", encoding="utf-8")
+        script = d / "run.sh"
+        script.write_text("#!/bin/sh\necho hi\n", encoding="utf-8")
+        script.chmod(0o755)
+
+        objects = ssc.ObjectSet()
+        tree_hash = ssc.build_tree(d, objects, max_object_bytes=ssc.DEFAULT_MAX_OBJECT_BYTES)
+        assert tree_hash.startswith("sha256:")
+        # tree object present and canonical
+        kind, data = objects.objects[tree_hash]
+        assert kind == ssc.KIND_TREE
+        tree = json.loads(data)
+        entries = {e["name"]: e for e in tree["entries"]}
+        assert entries["SKILL.md"]["mode"] == ssc.MODE_FILE
+        assert entries["run.sh"]["mode"] == ssc.MODE_EXEC
+        # entries sorted by name (byte order)
+        names = [e["name"] for e in tree["entries"]]
+        assert names == sorted(names)
+
+    def test_build_tree_dedups_identical_blobs(self, tmp_path):
+        d = tmp_path / "skill"
+        (d / "a").mkdir(parents=True)
+        (d / "b").mkdir(parents=True)
+        (d / "a" / "f.txt").write_text("same", encoding="utf-8")
+        (d / "b" / "f.txt").write_text("same", encoding="utf-8")
+        objects = ssc.ObjectSet()
+        ssc.build_tree(d, objects, max_object_bytes=ssc.DEFAULT_MAX_OBJECT_BYTES)
+        blob_hashes = [h for h, (k, _) in objects.objects.items() if k == ssc.KIND_BLOB]
+        # only one unique blob for the identical "same" content
+        assert len(set(blob_hashes)) == 1
+
+    def test_build_tree_skips_symlink(self, tmp_path):
+        d = tmp_path / "skill"
+        d.mkdir()
+        (d / "real.txt").write_text("x", encoding="utf-8")
+        try:
+            (d / "link.txt").symlink_to(d / "real.txt")
+        except (OSError, NotImplementedError):
+            pytest.skip("symlinks unsupported here")
+        objects = ssc.ObjectSet()
+        tree_hash = ssc.build_tree(d, objects, max_object_bytes=ssc.DEFAULT_MAX_OBJECT_BYTES)
+        tree = json.loads(objects.objects[tree_hash][1])
+        names = [e["name"] for e in tree["entries"]]
+        assert "link.txt" not in names
+        assert "real.txt" in names
+
+    def test_build_tree_rejects_oversize_blob(self, tmp_path):
+        d = tmp_path / "skill"
+        d.mkdir()
+        (d / "big").write_bytes(b"x" * 100)
+        objects = ssc.ObjectSet()
+        with pytest.raises(ValueError):
+            ssc.build_tree(d, objects, max_object_bytes=10)
+
+    def test_build_commit_shape(self):
+        objects = ssc.ObjectSet()
+        c = ssc.build_commit(
+            "sha256:tree", ["sha256:p"], owner="o", device="dev",
+            message="m", objects=objects, ts="2026-07-18T00:00:00Z",
+        )
+        commit = json.loads(objects.objects[c][1])
+        assert commit["type"] == "commit"
+        assert commit["tree"] == "sha256:tree"
+        assert commit["parents"] == ["sha256:p"]
+        assert commit["author"] == {"owner": "o", "device": "dev"}
+        assert commit["artifact_type"] == "skill"
+
+
+# ---------------------------------------------------------------------------
+# Three-way merge decision (contract §4.4, M1-C; mirrors skills_sync.py:619)
+# ---------------------------------------------------------------------------
+
+class TestMergeDecision:
+    def test_no_change(self):
+        assert ssc._merge_skill("b", "b", "b") == "either"
+
+    def test_ours_only_changed(self):
+        assert ssc._merge_skill("b", "o", "b") == "ours"
+
+    def test_theirs_only_changed(self):
+        assert ssc._merge_skill("b", "b", "t") == "theirs"
+
+    def test_both_converged(self):
+        assert ssc._merge_skill("b", "x", "x") == "either"
+
+    def test_true_overlap(self):
+        assert ssc._merge_skill("b", "o", "t") == "overlap"
+
+    def test_deleted_both(self):
+        assert ssc._merge_skill(None, None, None) == "none"
+
+
+# ---------------------------------------------------------------------------
+# End-to-end push / pull / conflict against the mock server
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def synced_env(tmp_path, monkeypatch):
+    """A HERMES_HOME with two opted-in skills + a token-carrying identity."""
+    import hermes_constants
+    home = tmp_path / "hermes"
+    skills = home / "skills"
+    skills.mkdir(parents=True)
+    monkeypatch.setattr(hermes_constants, "get_hermes_home", lambda: home)
+    monkeypatch.setattr(ssc, "_skills_dir", lambda: skills)
+
+    _write_skill(skills, "alpha", body="alpha v1\n")
+    _write_skill(skills, "beta", body="beta v1\n", category="devops")
+
+    # Opt both into sync + treat them as eligible (bypass bundled/hub checks).
+    monkeypatch.setattr(ssc, "list_synced_skill_names", lambda: ["alpha", "beta"])
+
+    def _rel(name):
+        from pathlib import PurePosixPath
+        return {"alpha": PurePosixPath("alpha"),
+                "beta": PurePosixPath("devops/beta")}.get(name)
+
+    monkeypatch.setattr(ssc, "_skill_rel_path", _rel)
+
+    def _find(name):
+        return {"alpha": skills / "alpha",
+                "beta": skills / "devops" / "beta"}.get(name)
+
+    import tools.skill_usage as su
+    monkeypatch.setattr(su, "_find_skill_dir", _find)
+
+    token = _jwt({"sub": "owner1", "tool_gateway_admin": True})
+    identity = {"api_key": token, "base_url": "http://x", "owner": "owner1",
+                "dev_gate_ok": True, "claims": {}}
+    return home, skills, identity
+
+
+class TestEndToEnd:
+    def test_capabilities_version_check(self, mock_server):
+        base, state = mock_server
+        client = ssc.HSPClient(base, "tok")
+        caps = client.capabilities()
+        assert caps["hsp_version"] == "1"
+        ssc._check_version(caps)  # no raise
+
+    def test_version_mismatch_raises(self, mock_server):
+        base, state = mock_server
+        state.hsp_version = "2"
+        client = ssc.HSPClient(base, "tok")
+        with pytest.raises(ssc.HSPError):
+            ssc._check_version(client.capabilities())
+
+    def test_push_uploads_and_cas(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+        result = ssc.push_skills(client, identity=identity)
+        assert result["ok"] is True
+        # HEAD ref advanced to our commit
+        head = state.refs["refs/user/owner1/HEAD"]
+        assert head == result["head"]
+        # commit object is present and well-formed
+        kind, data = state.objects[head]
+        assert kind == ssc.KIND_COMMIT
+        commit = json.loads(data)
+        assert commit["author"]["owner"] == "owner1"
+        assert commit["parents"] == []  # first commit
+
+    def test_push_then_pull_materializes(self, mock_server, synced_env, tmp_path, monkeypatch):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+        ssc.push_skills(client, identity=identity)
+
+        # Simulate a fresh device: new skills dir, same server, same opt-in.
+        dev2 = tmp_path / "hermes2" / "skills"
+        dev2.mkdir(parents=True)
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: dev2)
+        monkeypatch.setattr(ssc, "read_sync_state", lambda: {"head": None, "skills": {}})
+        saved = {}
+        monkeypatch.setattr(ssc, "write_sync_state", lambda d: saved.update(d))
+
+        result = ssc.pull_skills(client, identity=identity)
+        assert result["ok"] is True
+        assert "alpha" in result["updated"]
+        assert "devops/beta" in result["updated"]
+        # content materialized to disk
+        assert (dev2 / "alpha" / "SKILL.md").read_text().endswith("alpha v1\n")
+        assert (dev2 / "devops" / "beta" / "SKILL.md").read_text().endswith("beta v1\n")
+
+    def test_push_idempotent_reupload(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+        r1 = ssc.push_skills(client, identity=identity)
+        n_objects = len(state.objects)
+        # push again with no local change -> same head, objects already_present
+        r2 = ssc.push_skills(client, identity=identity)
+        assert r2["ok"] is True
+        assert r2["head"] == r1["head"]
+        assert len(state.objects) == n_objects  # nothing new stored
+
+    def test_conflict_nonoverlap_merges(self, mock_server, synced_env, monkeypatch):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+        # First push establishes a base head we record locally.
+        first = ssc.push_skills(client, identity=identity)
+        # Inject a divergent server head: change beta server-side so the next
+        # CAS loses. We simulate by forcing one 409 whose actual == current head
+        # (the server keeps the same tree, so no overlap on alpha which we edit).
+        (skills / "alpha" / "SKILL.md").write_text(
+            "---\nname: alpha\ndescription: test\n---\nalpha v2\n", encoding="utf-8"
+        )
+        state.force_conflict_once = True
+        result = ssc.push_skills(client, identity=identity)
+        # actual == our own head -> both-sides identical -> merge commit succeeds
+        assert result.get("ok") is True
+        assert result.get("merged") is True
+
+    def test_conflict_true_overlap_writes_conflict_ref(self, mock_server, synced_env, monkeypatch):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+        ssc.push_skills(client, identity=identity)
+
+        # Build a DIFFERENT server-side head for the SAME skill (alpha) so the
+        # three-way merge sees a true overlap. We construct it via a second
+        # snapshot after editing alpha differently, push it directly, then make
+        # our local head stale and edit alpha a third way.
+        (skills / "alpha" / "SKILL.md").write_text(
+            "---\nname: alpha\ndescription: test\n---\nSERVER edit\n", encoding="utf-8"
+        )
+        objs, root, _ = ssc.snapshot_profile(["alpha", "beta"])
+        their_commit = ssc.build_commit(
+            root, [], owner="owner1", device="other", message="theirs", objects=objs
+        )
+        client.put_objects(objs.objects)
+        state.refs["refs/user/owner1/HEAD"] = their_commit
+
+        # Our local edit to the same skill, from the OLD base -> true overlap.
+        (skills / "alpha" / "SKILL.md").write_text(
+            "---\nname: alpha\ndescription: test\n---\nLOCAL edit\n", encoding="utf-8"
+        )
+        result = ssc.push_skills(client, identity=identity)
+        assert result.get("conflict") is True
+        assert result["conflict_ref"].startswith("refs/user/owner1/conflict/")
+        assert "alpha" in result["overlapping_skills"]
+        # a conflict ref head was written server-side
+        assert result["conflict_ref"] in state.refs
+
+
+# ---------------------------------------------------------------------------
+# M1-D opt-in sidecar flag (tools/skill_usage.set_sync / is_sync_enabled)
+# ---------------------------------------------------------------------------
+
+class TestOptInFlag:
+    def test_set_and_read_sync_flag(self, tmp_path, monkeypatch):
+        import tools.skill_usage as su
+        monkeypatch.setattr(su, "_skills_dir", lambda: tmp_path)
+        # Make the skill curation-eligible so the gated mutator writes.
+        monkeypatch.setattr(su, "is_curation_eligible", lambda name, *a, **k: True)
+
+        assert su.is_sync_enabled("foo") is False
+        su.set_sync("foo", True)
+        assert su.is_sync_enabled("foo") is True
+        su.set_sync("foo", False)
+        assert su.is_sync_enabled("foo") is False
+
+    def test_sync_flag_ignored_for_ineligible(self, tmp_path, monkeypatch):
+        import tools.skill_usage as su
+        monkeypatch.setattr(su, "_skills_dir", lambda: tmp_path)
+        # Bundled/hub/external skills are not curation-eligible -> mutator no-ops.
+        monkeypatch.setattr(su, "is_curation_eligible", lambda name, *a, **k: False)
+        su.set_sync("bundled-skill", True)
+        assert su.is_sync_enabled("bundled-skill") is False
+
+
+# ---------------------------------------------------------------------------
+# §2.8 sync-manifest — opt-in as content in the sync plane (cross-device)
+# ---------------------------------------------------------------------------
+
+class TestSyncManifest:
+    def test_build_parse_roundtrip(self):
+        data = ssc.build_sync_manifest_bytes({"beta": True, "alpha": False})
+        parsed = ssc.parse_sync_manifest(data)
+        assert parsed == {"alpha": False, "beta": True}
+
+    def test_manifest_wire_shape(self):
+        # Must match gateway-gateway src/sync/manifest.ts: type + version:1 +
+        # skills:[{name,enabled}]. Skills sorted by name for a stable address.
+        import json
+        data = ssc.build_sync_manifest_bytes({"z": True, "a": True})
+        obj = json.loads(data.decode("utf-8"))
+        assert obj["type"] == "sync-manifest"
+        assert obj["version"] == 1
+        assert obj["skills"] == [
+            {"name": "a", "enabled": True},
+            {"name": "z", "enabled": True},
+        ]
+
+    def test_parse_rejects_malformed(self):
+        # Strict: unknown type, bad version, non-array skills, malformed entry.
+        assert ssc.parse_sync_manifest(b"not json") is None
+        assert ssc.parse_sync_manifest(b'{"type":"nope","version":1,"skills":[]}') is None
+        assert ssc.parse_sync_manifest(b'{"type":"sync-manifest","version":2,"skills":[]}') is None
+        assert ssc.parse_sync_manifest(b'{"type":"sync-manifest","version":1,"skills":{}}') is None
+        assert (
+            ssc.parse_sync_manifest(
+                b'{"type":"sync-manifest","version":1,"skills":[{"name":"x"}]}'
+            )
+            is None
+        )
+        # A malformed manifest must NOT be mistaken for "no skills opted in".
+        assert ssc.parse_sync_manifest(b'{"type":"sync-manifest","version":1,"skills":[]}') == {}
+
+    def test_snapshot_embeds_manifest_root_blob(self, mock_server, synced_env):
+        # snapshot_profile must add a root-level `sync-manifest` blob recording
+        # the opted-in set, alongside the skill subtrees, so opt-in is durable
+        # plane content. Read it back via read_manifest_of_root.
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+
+        objs, root_hash, skill_map = ssc.snapshot_profile(["alpha", "beta"])
+        client.put_objects(objs.objects)
+
+        manifest = ssc.read_manifest_of_root(client, root_hash)
+        assert manifest == {"alpha": True, "beta": True}
+
+        # The manifest is a root-level BLOB, not a skill subtree, so the skill
+        # walk must not surface it as a skill.
+        trees = ssc._skill_trees_of_root(client, root_hash)
+        assert "sync-manifest" not in trees
+        assert set(trees) == {"alpha", "devops/beta"}
+
+    def test_pull_adopts_opt_in_from_manifest(self, mock_server, synced_env, monkeypatch):
+        # A skill opted in on device A (present + enabled in the plane manifest)
+        # becomes opted in locally on pull, even if this device had it disabled.
+        base, state = mock_server
+        home, skills, identity = synced_env
+        client = ssc.HSPClient(base, identity["api_key"])
+
+        # Device A pushes alpha+beta (manifest enables both).
+        ssc.push_skills(client, identity=identity)
+
+        # Simulate device B: local opt-in intent is EMPTY, but eligibility passes.
+        adopted = {}
+        import tools.skill_usage as su
+        monkeypatch.setattr(su, "is_curation_eligible", lambda name, *a, **k: True)
+        monkeypatch.setattr(su, "is_sync_enabled", lambda name: False)
+        monkeypatch.setattr(su, "set_sync", lambda name, val: adopted.__setitem__(name, val))
+        # Local head unknown so the pull actually runs.
+        monkeypatch.setattr(ssc, "read_sync_state", lambda: {"head": None, "skills": {}})
+        monkeypatch.setattr(ssc, "write_sync_state", lambda d: None)
+        # No local opt-in gate (so materialize isn't the thing under test).
+        monkeypatch.setattr(ssc, "_opted_in_rel_paths", lambda: [])
+
+        result = ssc.pull_skills(client, identity=identity)
+        assert result["ok"] is True
+        # Both skills from the plane manifest were adopted into local opt-in.
+        assert adopted == {"alpha": True, "beta": True}
+        assert set(result["opt_in_adopted"]) == {"alpha", "beta"}
+
+
+# ---------------------------------------------------------------------------
+# Env-var configuration (Hermes Cloud "on by default" via environment)
+# ---------------------------------------------------------------------------
+
+class TestEnvConfig:
+    def test_base_url_env_wins(self, monkeypatch):
+        monkeypatch.setenv("HERMES_SYNC_BASE_URL", "https://plane.example/")
+        assert ssc.resolve_sync_base_url() == "https://plane.example"
+
+    def test_feature_enabled_env(self, monkeypatch):
+        # Default off.
+        monkeypatch.delenv("HERMES_SYNC_ENABLED", raising=False)
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {}, raising=False)
+        assert ssc.sync_feature_enabled() is False
+        for truthy in ("1", "true", "YES", "on"):
+            monkeypatch.setenv("HERMES_SYNC_ENABLED", truthy)
+            assert ssc.sync_feature_enabled() is True
+        for falsy in ("0", "false", "off"):
+            monkeypatch.setenv("HERMES_SYNC_ENABLED", falsy)
+            assert ssc.sync_feature_enabled() is False
+
+    def test_default_opt_in_env(self, monkeypatch):
+        monkeypatch.delenv("HERMES_SYNC_DEFAULT_OPT_IN", raising=False)
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {}, raising=False)
+        assert ssc.sync_default_opt_in() is False
+        monkeypatch.setenv("HERMES_SYNC_DEFAULT_OPT_IN", "true")
+        assert ssc.sync_default_opt_in() is True
+
+    def test_config_yaml_fallback_when_no_env(self, monkeypatch):
+        monkeypatch.delenv("HERMES_SYNC_ENABLED", raising=False)
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: {"sync": {"enabled": True}},
+            raising=False,
+        )
+        assert ssc.sync_feature_enabled() is True
+
+    def test_env_overrides_config_yaml(self, monkeypatch):
+        # Env wins over config.yaml (operator override precedence).
+        monkeypatch.setenv("HERMES_SYNC_ENABLED", "false")
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: {"sync": {"enabled": True}},
+            raising=False,
+        )
+        assert ssc.sync_feature_enabled() is False
+
+    def test_opt_out_policy_syncs_all_eligible(self, monkeypatch):
+        # With opt-out on, every eligible skill syncs even with no `sync:true`
+        # flag; an explicit `sync:false` still excludes.
+        monkeypatch.setattr(ssc, "sync_default_opt_in", lambda: True)
+        monkeypatch.setattr(ssc, "_all_local_skill_names", lambda: ["alpha", "beta", "gamma"])
+        monkeypatch.setattr(ssc, "is_sync_eligible", lambda n: n in {"alpha", "beta", "gamma"})
+        import tools.skill_usage as su
+        # gamma explicitly opted out; alpha/beta have no flag.
+        monkeypatch.setattr(su, "load_usage", lambda: {"gamma": {"sync": False}})
+        assert ssc.list_synced_skill_names() == ["alpha", "beta"]
+
+    def test_opt_in_policy_requires_flag(self, monkeypatch):
+        # With opt-out OFF (default opt-in), only explicitly-enabled skills sync.
+        monkeypatch.setattr(ssc, "sync_default_opt_in", lambda: False)
+        monkeypatch.setattr(ssc, "is_sync_eligible", lambda n: True)
+        import tools.skill_usage as su
+        monkeypatch.setattr(
+            su, "load_usage",
+            lambda: {"alpha": {"sync": True}, "beta": {}, "gamma": {"sync": False}},
+        )
+        assert ssc.list_synced_skill_names() == ["alpha"]
+
+
+class TestDeviceName:
+    def test_default_is_hostname_seeded(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: tmp_path)
+        monkeypatch.delenv("HERMES_SYNC_DEVICE_NAME", raising=False)
+        monkeypatch.setattr(
+            "socket.gethostname", lambda: "bens-macbook.local", raising=False
+        )
+        val = ssc.stable_device_id()
+        # short hostname + short suffix, NOT a bare 32-char hash
+        assert val.startswith("bens-macbook-")
+        assert val != "bens-macbook-"
+        # persisted + stable across calls
+        assert (tmp_path / ".sync_device_id").read_text() == val
+        assert ssc.stable_device_id() == val
+
+    def test_existing_file_wins_over_default_and_env(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: tmp_path)
+        (tmp_path / ".sync_device_id").write_text("Explicit Name", encoding="utf-8")
+        monkeypatch.setenv("HERMES_SYNC_DEVICE_NAME", "cloud-seed")
+        assert ssc.stable_device_id() == "Explicit Name"
+
+    def test_env_seeds_first_use(self, tmp_path, monkeypatch):
+        # Hermes Cloud path: HERMES_SYNC_DEVICE_NAME seeds the first-use label.
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: tmp_path)
+        monkeypatch.setenv("HERMES_SYNC_DEVICE_NAME", "hermes-cloud-ben-1")
+        assert ssc.stable_device_id() == "hermes-cloud-ben-1"
+        # persisted so it stays stable even if the env later changes
+        assert (tmp_path / ".sync_device_id").read_text() == "hermes-cloud-ben-1"
+        monkeypatch.setenv("HERMES_SYNC_DEVICE_NAME", "changed")
+        assert ssc.stable_device_id() == "hermes-cloud-ben-1"
+
+    def test_set_device_name_overwrites(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: tmp_path)
+        (tmp_path / ".sync_device_id").write_text("old", encoding="utf-8")
+        stored = ssc.set_device_name("  Ben's Laptop  ")
+        assert stored == "Ben's Laptop"  # trimmed
+        assert ssc.stable_device_id() == "Ben's Laptop"
+
+    def test_set_device_name_rejects_empty(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: tmp_path)
+        import pytest
+
+        with pytest.raises(ValueError):
+            ssc.set_device_name("   ")
+
+
+# ---------------------------------------------------------------------------
+# M2 org-shared skills (contract §11): identity gate, pull, propose (202/merge)
+# ---------------------------------------------------------------------------
+
+def _org_identity(role=None, org_id="org-1", owner="owner1"):
+    claims = {"sub": owner, "org_id": org_id, "tool_gateway_admin": True}
+    if role is not None:
+        claims["org_role"] = role
+    token = _jwt(claims)
+    return {"api_key": token, "base_url": "http://x", "owner": owner,
+            "dev_gate_ok": True, "claims": claims,
+            **({"org_id": org_id, "org_role": role} if role else {})}
+
+
+class TestOrgIdentityGate:
+    def test_org_identity_requires_role_claim(self, monkeypatch):
+        # Personal org: NAS stamps NO org_role -> inert, not an error path.
+        token = _jwt({"sub": "u", "org_id": "org-1"})
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token, "base_url": "https://x"})
+        with pytest.raises(ssc.SyncInertError):
+            ssc.resolve_org_identity()
+        assert ssc.org_sync_available() is False
+
+    def test_org_identity_with_role(self, monkeypatch):
+        token = _jwt({"sub": "u", "org_id": "org-9", "org_role": "MEMBER"})
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token, "base_url": "https://x"})
+        ident = ssc.resolve_org_identity()
+        assert ident["org_id"] == "org-9"
+        assert ident["org_role"] == "MEMBER"
+        assert ssc.org_sync_available() is True
+
+    def test_org_mirror_excluded_from_personal_sync(self, tmp_path, monkeypatch):
+        # A skill under _org/<id>/ must never be personal-sync eligible.
+        skills = tmp_path / "skills"
+        org_skill = skills / "_org" / "org-1" / "shared-x"
+        org_skill.mkdir(parents=True)
+        (org_skill / "SKILL.md").write_text("---\nname: shared-x\n---\n")
+        monkeypatch.setattr(ssc, "_skills_dir", lambda: skills)
+        import tools.skill_usage as su
+        monkeypatch.setattr(su, "is_bundled", lambda n: False)
+        monkeypatch.setattr(su, "is_hub_installed", lambda n: False)
+        monkeypatch.setattr(su, "_find_skill_dir", lambda n: org_skill)
+        import agent.skill_utils as sku
+        monkeypatch.setattr(sku, "is_external_skill_path", lambda p: False)
+        assert ssc.is_sync_eligible("shared-x") is False
+
+
+class TestOrgEndToEnd:
+    def test_admin_propose_merges_directly(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        identity = {**identity, "org_id": "org-1", "org_role": "ADMIN"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        result = ssc.propose_skill("alpha", client, identity=identity)
+        assert result["ok"] is True
+        assert result.get("merged") is True
+        head = state.refs["refs/org/org-1/HEAD"]
+        assert head == result["head"]
+        commit = json.loads(state.objects[head][1])
+        assert commit["parents"] == []  # first org commit
+
+    def test_member_propose_becomes_202_proposal(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        # Seed an org HEAD as admin first.
+        admin_ident = {**identity, "org_id": "org-1", "org_role": "ADMIN"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        seeded = ssc.propose_skill("alpha", client, identity=admin_ident)
+
+        # Member edits beta and proposes: server converts to 202.
+        state.org_role_admin = False
+        (skills / "devops" / "beta" / "SKILL.md").write_text(
+            "---\nname: beta\n---\nbeta v2 member edit\n", encoding="utf-8"
+        )
+        member_ident = {**identity, "org_id": "org-1", "org_role": "MEMBER"}
+        result = ssc.propose_skill("beta", client, identity=member_ident)
+        assert result["ok"] is True
+        assert result.get("proposal_pending") is True
+        assert result["proposal_id"] == 1
+        # HEAD untouched; proposal ref parked at the member's commit.
+        assert state.refs["refs/org/org-1/HEAD"] == seeded["head"]
+        assert state.refs["refs/org/org-1/proposals/1"] == result["commit"]
+        # NEVER reported as merged.
+        assert "merged" not in result
+
+    def test_member_proposal_splices_not_replaces(self, mock_server, synced_env):
+        # The proposed root must keep the OTHER skills from HEAD (per-skill
+        # delta, not a wholesale replace).
+        base, state = mock_server
+        home, skills, identity = synced_env
+        admin_ident = {**identity, "org_id": "org-1", "org_role": "ADMIN"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        ssc.propose_skill("alpha", client, identity=admin_ident)
+        ssc.propose_skill("beta", client, identity=admin_ident)
+
+        state.org_role_admin = False
+        member_ident = {**identity, "org_id": "org-1", "org_role": "MEMBER"}
+        result = ssc.propose_skill("alpha", client, identity=member_ident)
+        # Walk the proposed commit's root: both skills present.
+        commit = json.loads(state.objects[result["commit"]][1])
+        root = json.loads(state.objects[commit["tree"]][1])
+        names = {e["name"] for e in root["entries"]}
+        assert "alpha" in names and "devops" in names
+
+    def test_pull_org_skills_materializes_mirror(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        admin_ident = {**identity, "org_id": "org-1", "org_role": "ADMIN"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        ssc.propose_skill("alpha", client, identity=admin_ident)
+
+        result = ssc.pull_org_skills(client, identity=admin_ident)
+        assert result["ok"] is True
+        assert "alpha" in result["updated"]
+        mirrored = skills / "_org" / "org-1" / "alpha" / "SKILL.md"
+        assert mirrored.exists()
+        assert mirrored.read_text().endswith("alpha v1\n")
+
+    def test_pull_org_noop_when_no_head(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        ident = {**identity, "org_id": "org-1", "org_role": "MEMBER"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        result = ssc.pull_org_skills(client, identity=ident)
+        assert result["ok"] is True
+        assert result["head"] is None
+        assert result["updated"] == []
+
+    def test_propose_requires_org_feature(self, mock_server, synced_env):
+        base, state = mock_server
+        home, skills, identity = synced_env
+        state.org_feature = False
+        ident = {**identity, "org_id": "org-1", "org_role": "ADMIN"}
+        client = ssc.HSPClient(base, identity["api_key"])
+        with pytest.raises(ssc.SyncInertError):
+            ssc.propose_skill("alpha", client, identity=ident)
+
+    def test_maybe_pull_org_inert_without_role(self, monkeypatch):
+        # Personal org: no org_role claim -> None, never raises.
+        token = _jwt({"sub": "u", "org_id": "org-1"})
+        import hermes_cli.auth as auth_mod
+        monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
+                            lambda **kw: {"api_key": token})
+        assert ssc.maybe_pull_org_skills() is None

tools/skill_manager_tool.py

@@ -622,6 +622,34 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
     return None
 
 
+def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optional[Dict[str, Any]]:
+    """Refuse writes to org-mirror skills (M2, contract §11.11 / design §7.1).
+
+    The ``_org/`` mirror is materialized FROM the org HEAD and overwritten on
+    every pull — a local edit would be silently lost AND would misrepresent
+    admin-approved shared content. The change path is: fork into a personal
+    skill, edit, then ``hermes skills propose``.
+    """
+    try:
+        from agent.skill_utils import is_org_mirror_path
+
+        if is_org_mirror_path(skill_path, _skills_dir()):
+            return {
+                "success": False,
+                "error": (
+                    f"Refusing {action} for '{name}': it is an ORG-SHARED "
+                    "skill (read-only mirror of your org's approved set; "
+                    "local edits are overwritten on every org pull). To "
+                    "change it: copy it to a personal skill, edit that, then "
+                    "`hermes skills propose <name>` so an org admin can "
+                    "review and approve."
+                ),
+            }
+    except Exception:
+        logger.debug("org mirror guard lookup failed for %s", name, exc_info=True)
+    return None
+
+
 def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]:
     """Look for ``name`` under SKILL.md across OTHER Hermes profiles.
 
@@ -891,6 +919,9 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
     existing = _find_skill(name)
     if not existing:
         return {"success": False, "error": _skill_not_found_error(name)}
+    org_guard = _org_mirror_write_guard(name, existing["path"], "edit")
+    if org_guard:
+        return org_guard
     guard = _background_review_write_guard(name, existing["path"], "edit")
     if guard:
         return guard
@@ -953,6 +984,9 @@ def _patch_skill(
         return {"success": False, "error": _skill_not_found_error(name)}
 
     skill_dir = existing["path"]
+    org_guard = _org_mirror_write_guard(name, skill_dir, "patch")
+    if org_guard:
+        return org_guard
     guard = _background_review_write_guard(name, skill_dir, "patch")
     if guard:
         return guard
@@ -1059,6 +1093,9 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
     existing = _find_skill(name)
     if not existing:
         return {"success": False, "error": _skill_not_found_error(name)}
+    org_guard = _org_mirror_write_guard(name, existing["path"], "delete")
+    if org_guard:
+        return org_guard
     guard = _background_review_write_guard(name, existing["path"], "delete")
     if guard:
         return guard
@@ -1176,6 +1213,9 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
     existing = _find_skill(name)
     if not existing:
         return {"success": False, "error": _skill_not_found_error(name, " Create it first with action='create'.")}
+    org_guard = _org_mirror_write_guard(name, existing["path"], "write_file")
+    if org_guard:
+        return org_guard
     guard = _background_review_write_guard(name, existing["path"], "write_file")
     if guard:
         return guard
@@ -1337,6 +1377,56 @@ def apply_skill_pending(payload: Dict[str, Any]) -> str:
         _skill_gate_bypass.reset(token)
 
 
+# Debounce state for the HSP sync push hook. A burst of skill_manage writes
+# (e.g. create + several write_file calls) collapses into a single push after
+# a short quiet window, on a daemon timer so the agent write never blocks.
+_sync_push_timer = None
+_sync_push_lock = None
+_SYNC_PUSH_DEBOUNCE_S = 5.0
+
+
+def _maybe_debounced_sync_push(skill_name: str) -> None:
+    """Schedule a debounced best-effort HSP push after a skill write.
+
+    Cheap fast-path: if the skill isn't opted into sync, do nothing (no auth,
+    no network). Otherwise (re)arm a daemon timer; the actual push runs through
+    ``skills_sync_client.maybe_push_skills`` which enforces the DEV-PHASE gate
+    and swallows all errors. Never blocks the caller (M1-C: agent never blocks
+    on sync).
+    """
+    global _sync_push_timer, _sync_push_lock
+    try:
+        from tools.skill_usage import is_sync_enabled
+
+        if not is_sync_enabled(skill_name):
+            return
+    except Exception:
+        return
+
+    import threading
+
+    if _sync_push_lock is None:
+        _sync_push_lock = threading.Lock()
+
+    def _fire():
+        try:
+            from tools.skills_sync_client import maybe_push_skills
+
+            maybe_push_skills(message=f"sync: {skill_name}")
+        except Exception:
+            pass
+
+    with _sync_push_lock:
+        if _sync_push_timer is not None:
+            try:
+                _sync_push_timer.cancel()
+            except Exception:
+                pass
+        _sync_push_timer = threading.Timer(_SYNC_PUSH_DEBOUNCE_S, _fire)
+        _sync_push_timer.daemon = True
+        _sync_push_timer.start()
+
+
 def skill_manage(
     action: str,
     name: str,
@@ -1435,6 +1525,18 @@ def skill_manage(
         except Exception:
             pass
 
+        # HSP sync push hook (debounced, best-effort). Fires only AFTER the
+        # write gate passed (staged/unapproved writes never reach here -- the
+        # gate returns early above), so we never push un-reviewed content.
+        # Inert unless the DEV-PHASE gate is open (tool_gateway_admin on the
+        # token), a sync base URL is configured, and the skill is opted into
+        # sync. Debounced so a burst of edits collapses to one push. Never
+        # raises -- an agent write must never block on sync (M1-C invariant).
+        try:
+            _maybe_debounced_sync_push(name)
+        except Exception:
+            pass
+
     return json.dumps(result, ensure_ascii=False)
 
 

tools/skill_usage.py

@@ -450,12 +450,18 @@ def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) ->
     Agent-created skills are always eligible. Bundled built-ins become eligible
     only when ``curator.prune_builtins`` is enabled. Hub-installed and external
     skill-dir skills are NEVER eligible — they have an external upstream owner.
+    Org-mirror skills (``_org/``) are NEVER eligible — the org HEAD owns them;
+    curation happens via propose → approve, not local archive/consolidate.
     Protected built-ins (``PROTECTED_BUILTIN_SKILLS``) are NEVER eligible
     regardless of any flag — they back load-bearing UX and must never be
     archived or consolidated.
     """
+    from agent.skill_utils import is_org_mirror_path
+
     if skill_path is not None and is_external_skill_path(skill_path):
         return False
+    if skill_path is not None and is_org_mirror_path(skill_path, _skills_dir()):
+        return False
     if is_protected_builtin(skill_name):
         return False
     if is_hub_installed(skill_name):
@@ -464,6 +470,8 @@ def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) ->
         return _prune_builtins_enabled()
     local_dir = _find_skill_dir(skill_name)
     if local_dir is not None:
+        if is_org_mirror_path(local_dir, _skills_dir()):
+            return False
         return not is_external_skill_path(local_dir)
     if _find_external_skill_dir(skill_name) is not None:
         return False
@@ -675,6 +683,26 @@ def set_pinned(skill_name: str, pinned: bool) -> None:
     _mutate(skill_name, _apply, require_curation_eligible=True)
 
 
+def set_sync(skill_name: str, sync: bool) -> None:
+    """Set the HSP-sync opt-in flag on a skill's usage record (M1-D).
+
+    Sync is OPT-IN: nothing propagates to the sync plane unless the user marks
+    a skill with ``sync: true`` here. Sits alongside ``pinned``/``created_by``
+    on the ``.usage.json`` sidecar and is read by
+    ``tools.skills_sync_client.list_synced_skill_names``. Gated on curation
+    eligibility so bundled/hub/external skills (which never sync) can't be
+    marked. Provisional per the M1-D default.
+    """
+    def _apply(rec: Dict[str, Any]) -> None:
+        rec["sync"] = bool(sync)
+    _mutate(skill_name, _apply, require_curation_eligible=True)
+
+
+def is_sync_enabled(skill_name: str) -> bool:
+    """Whether a skill is opted into HSP sync (``sync: true`` in its record)."""
+    return get_record(skill_name).get("sync") is True
+
+
 def forget(skill_name: str) -> None:
     """Drop a skill's usage entry entirely. Called when the skill is deleted."""
     if not skill_name:
@@ -833,14 +861,16 @@ def _find_skill_dir(skill_name: str) -> Optional[Path]:
     """Locate the directory for a skill by its frontmatter `name:` field.
 
     Handles both flat (~/.hermes/skills/<skill>/SKILL.md) and category-nested
-    (~/.hermes/skills/<category>/<skill>/SKILL.md) layouts.
+    (~/.hermes/skills/<category>/<skill>/SKILL.md) layouts. Uses the gated
+    index iterator so M2 org mirrors resolve ONLY for the active org
+    (stale ``_org/<other>/`` trees never match).
     """
     base = _skills_dir()
     if not base.exists():
         return None
-    for skill_md in base.rglob("SKILL.md"):
-        if is_excluded_skill_path(skill_md):
-            continue
+    from agent.skill_utils import iter_skill_index_files
+
+    for skill_md in iter_skill_index_files(base, "SKILL.md"):
         if is_external_skill_path(skill_md):
             continue
         if _read_skill_name(skill_md, fallback=skill_md.parent.name) == skill_name:

tools/skills_tool.py

@@ -1561,6 +1561,69 @@ def skill_view(
                     "Could not preprocess skill content for %s", skill_name, exc_info=True
                 )
 
+        # ── M2 org provenance header (load-time) ──────────────────────────
+        # An org-shared skill announces its provenance IN the returned content
+        # — the moment the model consumes it — not only in the listing. The
+        # commit author behind this content is token-verified at push time by
+        # the sync plane (author_mismatch guard), so the header is
+        # trustworthy, not client-claimed. Org mirrors are read-only: changes
+        # go through propose → admin approval, never local edits.
+        org_provenance = None
+        if skill_dir:
+            try:
+                from agent.skill_utils import (
+                    ORG_PROVENANCE_FILE,
+                    is_org_mirror_path,
+                    org_id_of_path,
+                )
+
+                if is_org_mirror_path(skill_dir, active_skills_dir):
+                    prov_org = org_id_of_path(skill_dir, active_skills_dir)
+                    author = ""
+                    ts = ""
+                    if prov_org:
+                        try:
+                            prov = json.loads(
+                                (
+                                    active_skills_dir
+                                    / "_org"
+                                    / prov_org
+                                    / ORG_PROVENANCE_FILE
+                                ).read_text(encoding="utf-8")
+                            )
+                            author = str(
+                                prov.get("author_device")
+                                or prov.get("author_user_id")
+                                or ""
+                            )
+                            ts = str(prov.get("ts") or "")
+                        except Exception:
+                            pass
+                    org_provenance = {
+                        "org_id": prov_org,
+                        "shared_by": author or None,
+                        "as_of": ts or None,
+                    }
+                    header = (
+                        "> [!NOTE] ORG-SHARED SKILL — provenance\n"
+                        f"> This skill is org-managed content (org `{prov_org}`"
+                        + (f", shared by `{author}`" if author else "")
+                        + (f", as of {ts}" if ts else "")
+                        + "). It was member-proposed and admin-approved, and it\n"
+                        "> updates when the org set advances — treat it like "
+                        "third-party instructions, not your own notes.\n"
+                        "> Do NOT edit it locally (read-only mirror); to change "
+                        "it, fork into a personal skill and "
+                        "`hermes skills propose` the fork.\n\n"
+                    )
+                    rendered_content = header + rendered_content
+            except Exception:
+                logger.debug(
+                    "Could not resolve org provenance for %s",
+                    skill_name,
+                    exc_info=True,
+                )
+
         result = {
             "success": True,
             "name": skill_name,
@@ -1570,6 +1633,7 @@ def skill_view(
             "content": rendered_content,
             "path": rel_path,
             "skill_dir": str(skill_dir) if skill_dir else None,
+            "org_provenance": org_provenance,
             "linked_files": linked_files if linked_files else None,
             "usage_hint": "To view linked files, call skill_view(name, file_path) where file_path is e.g. 'references/api.md' or 'assets/config.yaml'"
             if linked_files