diff --git a/cli.py b/cli.py index 6cb6345f4be..2e10fcd24a3 100644 --- a/cli.py +++ b/cli.py @@ -13518,6 +13518,16 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): ) 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( diff --git a/gateway/run.py b/gateway/run.py index 61fcd96eb70..a9c6dce8b4c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -21145,6 +21145,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") diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 1525041c111..91d77519d78 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -282,6 +282,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 @@ -4304,6 +4305,103 @@ 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 \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 Opt a skill into sync (M1-D opt-in)\n" + " disable Opt a skill out of sync", + file=sys.stderr, + ) + return 1 + + 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("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 @@ -13464,6 +13562,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) diff --git a/hermes_cli/subcommands/sync.py b/hermes_cli/subcommands/sync.py new file mode 100644 index 00000000000..6473d74588a --- /dev/null +++ b/hermes_cli/subcommands/sync.py @@ -0,0 +1,44 @@ +"""``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 -- opt a skill into sync (M1-D) + hermes sync disable -- opt a skill out of sync + +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)") + + sync_parser.set_defaults(func=cmd_sync) diff --git a/tests/tools/test_skills_sync_client.py b/tests/tools/test_skills_sync_client.py new file mode 100644 index 00000000000..b9bd2571590 --- /dev/null +++ b/tests/tools/test_skills_sync_client.py @@ -0,0 +1,582 @@ +"""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 + + +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": + return self._json(200, { + "hsp_version": state.hsp_version, + "features": ["personal"], + "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"" + + if self.path == "/v1/sync/objects": + return self._handle_put_objects(raw) + + if self.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") + 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_manifest", lambda: {"head": None, "skills": {}}) + saved = {} + monkeypatch.setattr(ssc, "write_sync_manifest", 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 diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index eaf30dd41ad..f8fd8200a22 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -1320,6 +1320,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, @@ -1418,6 +1468,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) diff --git a/tools/skill_usage.py b/tools/skill_usage.py index dcdca87f812..b38f64f2afb 100644 --- a/tools/skill_usage.py +++ b/tools/skill_usage.py @@ -675,6 +675,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: diff --git a/tools/skills_sync_client.py b/tools/skills_sync_client.py new file mode 100644 index 00000000000..d22da3d1756 --- /dev/null +++ b/tools/skills_sync_client.py @@ -0,0 +1,1150 @@ +#!/usr/bin/env python3 +""" +HSP/1 sync client -- Hermes Sync Protocol version 1, client (personal skill sync). + +This is the LOW-LEVEL sync layer. It builds content-addressed HSP objects +(blob/tree/commit) from local skills, talks the HSP/1 wire contract to a sync +plane (push objects + CAS a ref, pull the owner's HEAD, three-way merge on a +409), and is driven by: + + * a debounced push hook in ``skill_manage`` (after the write-gate passes), + * a periodic pull hook (``maybe_pull_skills``) at the curator tick sites, + * the ``hermes sync status|pull|push|now`` CLI. + +It lives beside ``tools/skills_sync.py`` (NOT under ``hermes_cli/``) so the +low-level sync layer never imports the CLI -- same rule the bundled-skills +sync module documents at ``skills_sync.py:43-50``. + +Contract: ``~/src/specs/collective-wisdom/hsp-1-contract.md`` (HSP/1, frozen +for Milestone 1). Endpoint shapes, object model, canonicalization, and status +codes below all trace to that document. + +--- DEV-PHASE GATE (Milestone 1) ----------------------------------------- +Client sync is INERT (no push, no pull, no-op) unless the resolved Nous +identity's access token carries ``tool_gateway_admin === true``. That claim is +minted by NAS (access-token-issuer.ts:312) and rides on the same bearer +``resolve_nous_runtime_credentials()`` returns. We decode the JWT payload +(no signature verification -- the server re-verifies) and check the claim +before doing any sync work. This is a temporary dev gate for the M1 rollout; +remove it (or replace it with a real ``sync:*`` scope / config toggle) when +sync ships to all users. + +--- OPT-IN DEFAULT (M1-D, provisional) ----------------------------------- +Nothing syncs unless the user marks a skill for sync via a ``sync`` flag on +the skill's usage sidecar (alongside ``pinned``/``created_by`` in +``.usage.json``). Only agent-created + user-authored skills under +``~/.hermes/skills/`` are eligible; bundled and hub-installed skills are +excluded. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import time +import stat as _stat +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Any, Callable, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# HSP/1 protocol constants (contract §1, §3.1) +HSP_VERSION = "1" +DEFAULT_MAX_OBJECT_BYTES = 26214400 # 25 MiB, mirrors capabilities default + +# Object kinds (contract §2) +KIND_BLOB = "blob" +KIND_TREE = "tree" +KIND_COMMIT = "commit" + +# Tree entry modes (contract §2.3) +MODE_FILE = "file" +MODE_EXEC = "exec" +MODE_DIR = "dir" + +ARTIFACT_TYPE_SKILL = "skill" + + +# --------------------------------------------------------------------------- +# Content addressing (contract §2.1 / OI-5) +# +# HSP uses the FULL 64-hex sha256 digest on the wire. This is a DIFFERENT +# namespace from hermes-agent's local ``content_hash`` (skills_guard.py:846), +# which is a truncated 16-hex digest used for local dedup. They must never be +# conflated -- we compute full digests here. +# --------------------------------------------------------------------------- + +def hsp_address(data: bytes) -> str: + """Return ``sha256:<64-hex>`` -- the HSP wire address of ``data`` (contract §2.1).""" + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def canonical_json_bytes(obj: Dict[str, Any]) -> bytes: + """Canonical JSON serialization for tree/commit hashing (contract §2.5). + + UTF-8, keys sorted lexicographically, no insignificant whitespace + (``separators=(",", ":")``), no trailing newline. Arrays must already be + in the contract-specified order by the caller (tree entries by ``name``, + commit ``parents`` in significance order). Both client and server MUST + produce byte-identical output or a push fails ``422 hash_mismatch``. + """ + return json.dumps( + obj, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + + +# --------------------------------------------------------------------------- +# Identity & DEV-PHASE gate +# +# We reuse resolve_nous_runtime_credentials() for the bearer (it honors the +# cross-process file lock + portal host allowlist and refreshes as needed -- +# we do NOT reimplement refresh). The returned api_key IS the JWT bearer; we +# decode its payload (unverified) to read the dev gate claim. +# --------------------------------------------------------------------------- + +# Dev-phase gate claim (NAS access-token-issuer.ts:312). Sync is inert unless +# the resolved token carries this claim === true. Remove when sync ships GA. +DEV_GATE_CLAIM = "tool_gateway_admin" + + +class SyncInertError(RuntimeError): + """Raised (and caught by the gate-and-swallow hooks) when sync must no-op: + + not logged in, no bearer, or the dev-phase gate claim is absent/false. + """ + + +def _decode_jwt_payload_unverified(token: str) -> Dict[str, Any]: + """Decode a JWT payload WITHOUT signature verification. + + Safe here: we never trust these claims for authz -- the server re-verifies + every call. We only read the dev-gate claim to decide whether to attempt + sync at all. Mirrors the diagnostic decode in + plugins/dashboard_auth/nous/__init__.py:463. + """ + try: + import jwt # PyJWT, a core dependency + + return jwt.decode( + token, + options={"verify_signature": False, "verify_exp": False}, + ) or {} + except Exception as e: + logger.debug("skills_sync_client: JWT payload decode failed: %s", e) + return {} + + +def resolve_identity() -> Dict[str, Any]: + """Resolve the Nous bearer + owner + dev-gate flag. + + Returns a dict: ``{api_key, base_url, owner, dev_gate_ok, claims}``. + Raises :class:`SyncInertError` if not logged in / no bearer. + + ``owner`` is the token-verified subject; the server derives the real owner + from the bearer regardless (contract §0.4), so this is advisory for local + ref naming only. + """ + try: + from hermes_cli.auth import resolve_nous_runtime_credentials + + creds = resolve_nous_runtime_credentials() + except Exception as e: + raise SyncInertError(f"no Nous credentials: {e}") from e + + api_key = (creds or {}).get("api_key") + if not api_key: + raise SyncInertError("no bearer token available") + + claims = _decode_jwt_payload_unverified(api_key) + owner = ( + claims.get("sub") + or claims.get("privy_did") + or claims.get("tid") + or "unknown" + ) + dev_gate_ok = claims.get(DEV_GATE_CLAIM) is True + return { + "api_key": api_key, + "base_url": (creds or {}).get("base_url"), + "owner": str(owner), + "dev_gate_ok": dev_gate_ok, + "claims": claims, + } + + +def dev_gate_open() -> bool: + """Whether the DEV-PHASE gate permits sync. Never raises.""" + try: + return bool(resolve_identity().get("dev_gate_ok")) + except SyncInertError: + return False + except Exception as e: + logger.debug("skills_sync_client: dev_gate_open check failed: %s", e) + return False + + +# --------------------------------------------------------------------------- +# Sync-plane endpoint resolution +# +# The HSP routes are mounted under /v1/sync/ (contract §1). The base URL is +# configurable (config.yaml sync.base_url or HERMES_SYNC_BASE_URL bridge env); +# it is NOT the inference base_url. When unset, sync is inert -- there is no +# server to talk to yet (the server is being built in parallel). +# --------------------------------------------------------------------------- + +def resolve_sync_base_url() -> Optional[str]: + """Resolve the HSP sync-plane base URL, or None when unconfigured. + + Order: HERMES_SYNC_BASE_URL env bridge -> config.yaml ``sync.base_url``. + Returns a base without a trailing slash (e.g. ``https://host``); the + ``/v1/sync/`` prefix is appended by the client. + """ + env = os.getenv("HERMES_SYNC_BASE_URL") + if env and env.strip(): + return env.strip().rstrip("/") + try: + # Lazy import: the low-level sync layer must not import the CLI at + # module load (skills_sync.py:43-50). A function-scoped import avoids + # the cycle -- same pattern agent/curator.py:141 uses for config. + from hermes_cli.config import load_config + + cfg = load_config() or {} + sync_cfg = cfg.get("sync") or {} + base = sync_cfg.get("base_url") + if isinstance(base, str) and base.strip(): + return base.strip().rstrip("/") + except Exception as e: + logger.debug("skills_sync_client: config sync.base_url read failed: %s", e) + return None + + +# --------------------------------------------------------------------------- +# Local skill eligibility + the M1-D opt-in "sync" flag +# +# Only agent-created + user-authored skills under ~/.hermes/skills/ sync. +# Bundled (.bundled_manifest) and hub-installed skills are excluded. Sync is +# opt-in: a skill only syncs when its usage-sidecar carries ``sync: true``. +# --------------------------------------------------------------------------- + +def _skills_dir() -> Path: + from hermes_constants import get_hermes_home + + return get_hermes_home() / "skills" + + +def is_sync_eligible(skill_name: str) -> bool: + """Whether *skill_name* is a candidate for HSP sync (before the opt-in check). + + Eligible = present locally under ~/.hermes/skills/, NOT bundled, NOT + hub-installed, NOT an external-dir skill. Mirrors the exclusion logic used + by the curator (tools/skill_usage.py). + """ + try: + from tools.skill_usage import is_bundled, is_hub_installed, _find_skill_dir + from agent.skill_utils import is_external_skill_path + except Exception: + return False + if is_bundled(skill_name) or is_hub_installed(skill_name): + return False + skill_dir = _find_skill_dir(skill_name) + if skill_dir is None: + return False + if is_external_skill_path(skill_dir): + return False + return True + + +def list_synced_skill_names() -> List[str]: + """Return the names of skills the user has opted into sync (``sync: true``) + AND that remain eligible. Sorted, deduped.""" + try: + from tools.skill_usage import load_usage + except Exception: + return [] + names = [] + for name, rec in (load_usage() or {}).items(): + if isinstance(rec, dict) and rec.get("sync") is True and is_sync_eligible(name): + names.append(name) + return sorted(set(names)) + + +# --------------------------------------------------------------------------- +# Object building -- turn a skill directory into HSP blob/tree/commit objects +# +# A skill dir becomes one tree (contract §2.3). Each file is a blob; each +# subdir a nested tree. The profile-root tree (contract §2.3: "a tree whose +# entries are category trees") is built from the set of synced skill trees. +# --------------------------------------------------------------------------- + +class ObjectSet: + """Accumulates HSP objects to push: hash -> (kind, bytes). + + Deduped by content address, so identical blobs across skills upload once. + """ + + def __init__(self) -> None: + self.objects: Dict[str, Tuple[str, bytes]] = {} + + def add(self, kind: str, data: bytes) -> str: + addr = hsp_address(data) + self.objects.setdefault(addr, (kind, data)) + return addr + + def __len__(self) -> int: + return len(self.objects) + + +def _file_mode(path: Path) -> str: + """Return the HSP tree mode for a regular file: ``exec`` if +x else ``file`` + (contract §2.3). No symlinks / other modes are emitted.""" + try: + if path.stat().st_mode & (_stat.S_IXUSR | _stat.S_IXGRP | _stat.S_IXOTH): + return MODE_EXEC + except OSError: + pass + return MODE_FILE + + +def build_tree(dir_path: Path, objects: ObjectSet, *, max_object_bytes: int) -> str: + """Recursively build HSP objects for *dir_path*; return the tree address. + + Regular files become blobs; subdirectories become nested trees. Symlinks, + sockets, and other special files are skipped (contract §2.3 security: no + symlinks). Blobs over *max_object_bytes* raise :class:`ValueError` so the + caller can surface / skip the artifact (contract §4.3 -> 413). + """ + entries: List[Dict[str, str]] = [] + for child in sorted(dir_path.iterdir(), key=lambda p: p.name): + if child.is_symlink(): + logger.debug("skills_sync_client: skipping symlink %s", child) + continue + if child.is_dir(): + sub_hash = build_tree(child, objects, max_object_bytes=max_object_bytes) + entries.append( + {"name": child.name, "kind": KIND_TREE, "hash": sub_hash, "mode": MODE_DIR} + ) + elif child.is_file(): + data = child.read_bytes() + if len(data) > max_object_bytes: + raise ValueError( + f"file {child} is {len(data)} bytes > max_object_bytes " + f"{max_object_bytes} (contract §4.3)" + ) + blob_hash = objects.add(KIND_BLOB, data) + entries.append( + { + "name": child.name, + "kind": KIND_BLOB, + "hash": blob_hash, + "mode": _file_mode(child), + } + ) + # else: skip special files + # Entries sorted by name (byte order) for canonicalization (contract §2.3). + entries.sort(key=lambda e: e["name"]) + tree_obj = {"type": KIND_TREE, "entries": entries} + return objects.add(KIND_TREE, canonical_json_bytes(tree_obj)) + + +def build_commit( + tree_hash: str, + parents: List[str], + *, + owner: str, + device: str, + message: str, + objects: ObjectSet, + ts: Optional[str] = None, +) -> str: + """Build a commit object (contract §2.4) and return its address. + + ``parents``: 0 for first commit, 1 for a normal edit, 2 for a merge commit + (order significant: parents[0] = base fast-forwarded from, parents[1] = + the other head being merged). + """ + commit_obj = { + "type": KIND_COMMIT, + "tree": tree_hash, + "parents": list(parents), + "author": {"owner": owner, "device": device}, + "ts": ts or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "message": message, + "artifact_type": ARTIFACT_TYPE_SKILL, + } + return objects.add(KIND_COMMIT, canonical_json_bytes(commit_obj)) + + +def stable_device_id() -> str: + """Return an opaque, stable per-device id for commit ``author.device`` + (contract §2.4 -- advisory, never an auth input). Persisted under + ~/.hermes/skills/.sync_device_id.""" + path = _skills_dir() / ".sync_device_id" + try: + if path.exists(): + val = path.read_text(encoding="utf-8").strip() + if val: + return val + except OSError: + pass + import uuid + + val = uuid.uuid4().hex + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(val, encoding="utf-8") + except OSError as e: + logger.debug("skills_sync_client: could not persist device id: %s", e) + return val + + +# --------------------------------------------------------------------------- +# HSP/1 wire client +# +# Thin requests-based client for the endpoints in contract §3-§4. Uploads all +# new objects (batch), then CAS-es the ref. A 409 returns the actual head for +# the caller's three-way merge. Auth is the Nous bearer resolved above. +# --------------------------------------------------------------------------- + +class HSPError(RuntimeError): + """A non-recoverable HSP wire error (4xx that the client can't retry).""" + + def __init__(self, message: str, *, status: Optional[int] = None): + super().__init__(message) + self.status = status + + +class HSPConflict(RuntimeError): + """CAS lost (409). ``actual`` is the current head to merge against + (contract §4.4). NOT a rejection -- pushed objects are already durable.""" + + def __init__(self, actual: str): + super().__init__(f"CAS conflict; actual head {actual}") + self.actual = actual + + +class HSPClient: + """HSP/1 client bound to a base URL + bearer (contract §1, routes under + ``/v1/sync/``).""" + + def __init__(self, base_url: str, api_key: str, *, timeout: float = 30.0): + self.base = base_url.rstrip("/") + self.api_key = api_key + self.timeout = timeout + import requests # core dependency + + self._session = requests.Session() + self._session.headers["Authorization"] = f"Bearer {api_key}" + + def _url(self, path: str) -> str: + return f"{self.base}/v1/sync/{path.lstrip('/')}" + + # -- capability & read ------------------------------------------------- + + def capabilities(self) -> Dict[str, Any]: + """GET /v1/sync/capabilities (contract §3.1). No auth required.""" + r = self._session.get(self._url("capabilities"), timeout=self.timeout) + if r.status_code != 200: + raise HSPError(f"capabilities failed: {r.status_code}", status=r.status_code) + return r.json() + + def get_refs(self, prefix: str) -> List[Dict[str, str]]: + """GET /v1/sync/refs?prefix=... (contract §3.2).""" + r = self._session.get( + self._url("refs"), params={"prefix": prefix}, timeout=self.timeout + ) + if r.status_code != 200: + raise HSPError(f"get_refs failed: {r.status_code}", status=r.status_code) + return (r.json() or {}).get("refs", []) + + def get_object(self, obj_hash: str) -> Tuple[str, bytes]: + """GET /v1/sync/objects/:hash (contract §3.3). Returns (kind, bytes). + + Kind comes from ``X-HSP-Object-Type`` for tree/commit; a blob response + (application/octet-stream) is returned as ``blob``. + """ + r = self._session.get(self._url(f"objects/{obj_hash}"), timeout=self.timeout) + if r.status_code == 404: + raise HSPError(f"object {obj_hash} not found", status=404) + if r.status_code == 403: + raise HSPError(f"object {obj_hash} not readable", status=403) + if r.status_code != 200: + raise HSPError(f"get_object failed: {r.status_code}", status=r.status_code) + kind = r.headers.get("X-HSP-Object-Type") or KIND_BLOB + return kind, r.content + + def get_commit_json(self, commit_hash: str) -> Dict[str, Any]: + """Fetch a commit object and parse its canonical JSON.""" + kind, data = self.get_object(commit_hash) + if kind != KIND_COMMIT: + raise HSPError(f"{commit_hash} is {kind}, expected commit") + return json.loads(data.decode("utf-8")) + + def get_tree_json(self, tree_hash: str) -> Dict[str, Any]: + """Fetch a tree object and parse its canonical JSON.""" + kind, data = self.get_object(tree_hash) + if kind != KIND_TREE: + raise HSPError(f"{tree_hash} is {kind}, expected tree") + return json.loads(data.decode("utf-8")) + + # -- write ------------------------------------------------------------- + + def put_objects(self, objects: Dict[str, Tuple[str, bytes]]) -> Dict[str, Any]: + """POST /v1/sync/objects (contract §4.2). Batch multi-object upload. + + Contract §1 requires raw object bytes on the wire (NOT base64-in-JSON), + and §4.2 specifies "a length-prefixed or multipart stream of + {hash, type, bytes}". We use multipart/form-data: one part per object, + the part's field name = the claimed ``sha256:`` hash, its + ``filename`` carries the object ``type`` (blob|tree|commit), and the + part body is the raw object bytes. The server recomputes each hash from + the received bytes and rejects the whole batch with 422 on mismatch. + Idempotent: a known hash is a no-op ``already_present``. + + NOTE (framing choice within contract latitude): §4.2 says "length- + prefixed OR multipart"; this picks multipart/form-data with + (field=hash, filename=type, body=raw-bytes). The server strand must + parse the same framing -- flagged for cross-strand alignment. + """ + # (field_name, (filename, raw_bytes, content_type)) + files = [ + (h, (kind, data, "application/octet-stream")) + for h, (kind, data) in objects.items() + ] + r = self._session.post( + self._url("objects"), files=files, timeout=self.timeout + ) + if r.status_code == 413: + raise HSPError("object too large (413)", status=413) + if r.status_code == 422: + raise HSPError(f"hash_mismatch (422): {r.text}", status=422) + if r.status_code not in (200, 201): + raise HSPError(f"put_objects failed: {r.status_code}", status=r.status_code) + return r.json() if r.content else {} + + def cas_ref(self, name: str, from_hash: Optional[str], to_hash: str) -> Dict[str, Any]: + """POST /v1/sync/refs/:name -- atomic compare-and-swap (contract §4.4). + + Raises :class:`HSPConflict` (carrying the actual head) on 409. + """ + r = self._session.post( + self._url(f"refs/{name}"), + json={"from": from_hash, "to": to_hash}, + timeout=self.timeout, + ) + if r.status_code == 409: + actual = (r.json() or {}).get("actual", "") + raise HSPConflict(actual) + if r.status_code == 403: + raise HSPError("forbidden (403) -- owner/permission", status=403) + if r.status_code != 200: + raise HSPError(f"cas_ref failed: {r.status_code}", status=r.status_code) + return r.json() if r.content else {} + + +# --------------------------------------------------------------------------- +# HSP sync manifest (client-local, FULL-digest namespace) +# +# Records, per synced skill, the last commit HEAD we pushed/pulled and the +# tree hash of the on-disk content at that point. Distinct from the bundled +# manifest (skills_sync.py, truncated local content_hash namespace). Lives at +# ~/.hermes/skills/.sync_manifest as JSON. +# --------------------------------------------------------------------------- + +def _sync_manifest_path() -> Path: + return _skills_dir() / ".sync_manifest" + + +def read_sync_manifest() -> Dict[str, Any]: + """Read the HSP sync manifest. Returns {} on missing/corrupt. + + Shape: ``{"head": "sha256:...|null", "skills": {name: {tree, commit}}}``. + ``head`` is the last profile-root HEAD commit we reconciled with. + """ + path = _sync_manifest_path() + if not path.exists(): + return {"head": None, "skills": {}} + try: + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + data.setdefault("head", None) + data.setdefault("skills", {}) + return data + except (OSError, json.JSONDecodeError) as e: + logger.debug("skills_sync_client: sync manifest read failed: %s", e) + return {"head": None, "skills": {}} + + +def write_sync_manifest(data: Dict[str, Any]) -> None: + """Write the HSP sync manifest atomically. Best-effort.""" + import tempfile + + path = _sync_manifest_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".sync_manifest_", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + except Exception as e: + logger.debug("skills_sync_client: sync manifest write failed: %s", e) + + +# --------------------------------------------------------------------------- +# Tree materialization (pull) -- write an HSP tree back to a skill directory +# --------------------------------------------------------------------------- + +def materialize_tree(client: HSPClient, tree_hash: str, dest: Path) -> None: + """Write the HSP tree at *tree_hash* into *dest* (created if needed). + + Blobs become files (with +x restored for ``exec`` mode), nested trees + become subdirectories. Does NOT delete files absent from the tree -- the + caller decides removal semantics. Refuses path traversal via entry names. + """ + dest.mkdir(parents=True, exist_ok=True) + tree = client.get_tree_json(tree_hash) + for entry in tree.get("entries", []): + name = entry.get("name", "") + if not name or "/" in name or name in (".", ".."): + logger.warning("skills_sync_client: skipping unsafe tree entry %r", name) + continue + target = dest / name + kind = entry.get("kind") + if kind == KIND_TREE: + materialize_tree(client, entry["hash"], target) + elif kind == KIND_BLOB: + _, data = client.get_object(entry["hash"]) + target.write_bytes(data) + if entry.get("mode") == MODE_EXEC: + try: + st = target.stat().st_mode + target.chmod(st | _stat.S_IXUSR | _stat.S_IXGRP | _stat.S_IXOTH) + except OSError: + pass + + +# --------------------------------------------------------------------------- +# Profile snapshot -- build the objects + per-skill tree map for a push +# +# The profile root is a tree whose entries mirror each synced skill's relative +# path under ~/.hermes/skills/ (contract §2.3: "the profile root is a tree +# whose entries are category trees"). Only opted-in, eligible skills are +# included (M1-D opt-in + eligibility). +# --------------------------------------------------------------------------- + +def _skill_rel_path(skill_name: str) -> Optional[PurePosixPath]: + """Return the skill's path relative to ~/.hermes/skills/ (posix), or None.""" + try: + from tools.skill_usage import _find_skill_dir + except Exception: + return None + skill_dir = _find_skill_dir(skill_name) + if skill_dir is None: + return None + try: + rel = skill_dir.resolve().relative_to(_skills_dir().resolve()) + except (OSError, ValueError): + return None + return PurePosixPath(rel.as_posix()) + + +def snapshot_profile( + skill_names: List[str], *, max_object_bytes: int = DEFAULT_MAX_OBJECT_BYTES +) -> Tuple[ObjectSet, str, Dict[str, str]]: + """Build all HSP objects for *skill_names* + the profile-root tree. + + Returns ``(objects, root_tree_hash, skill_tree_map)`` where + ``skill_tree_map`` is ``{skill_name: tree_hash}``. Skills whose blobs + exceed *max_object_bytes* are skipped (surfaced via logger). + + The root tree nests category directories: a skill at ``devops/foo`` yields + a root entry ``devops`` (tree) containing ``foo`` (tree). Flat skills yield + a direct root entry. + """ + from tools.skill_usage import _find_skill_dir + + objects = ObjectSet() + skill_tree_map: Dict[str, str] = {} + # Nested dict representing the root: {name: {"__tree__": hash} | subdict} + root: Dict[str, Any] = {} + + for name in sorted(set(skill_names)): + rel = _skill_rel_path(name) + skill_dir = _find_skill_dir(name) + if rel is None or skill_dir is None: + continue + try: + tree_hash = build_tree(skill_dir, objects, max_object_bytes=max_object_bytes) + except ValueError as e: + logger.warning("skills_sync_client: skipping %s: %s", name, e) + continue + skill_tree_map[name] = tree_hash + # Insert into the nested root structure by relative path parts. + parts = list(rel.parts) + node = root + for part in parts[:-1]: + node = node.setdefault(part, {}) + node[parts[-1]] = {"__tree__": tree_hash} + + root_hash = _build_root_tree(root, objects) + return objects, root_hash, skill_tree_map + + +def _build_root_tree(node: Dict[str, Any], objects: ObjectSet) -> str: + """Recursively canonicalize the nested root structure into HSP trees.""" + entries: List[Dict[str, str]] = [] + for name, child in node.items(): + if isinstance(child, dict) and "__tree__" in child and len(child) == 1: + entries.append( + {"name": name, "kind": KIND_TREE, "hash": child["__tree__"], "mode": MODE_DIR} + ) + else: + sub_hash = _build_root_tree(child, objects) + entries.append( + {"name": name, "kind": KIND_TREE, "hash": sub_hash, "mode": MODE_DIR} + ) + entries.sort(key=lambda e: e["name"]) + tree_obj = {"type": KIND_TREE, "entries": entries} + return objects.add(KIND_TREE, canonical_json_bytes(tree_obj)) + + +# --------------------------------------------------------------------------- +# Ref naming (contract §2.6) +# --------------------------------------------------------------------------- + +def user_head_ref(owner: str) -> str: + return f"refs/user/{owner}/HEAD" + + +def user_conflict_ref(owner: str, n: int) -> str: + return f"refs/user/{owner}/conflict/{n}" + + +def _root_tree_of_commit(client: "HSPClient", commit_hash: str) -> str: + """Return the tree hash referenced by a commit.""" + return client.get_commit_json(commit_hash)["tree"] + + +def _skill_trees_of_root(client: "HSPClient", root_tree_hash: str) -> Dict[str, str]: + """Flatten a profile-root tree into ``{posix_rel_path: skill_tree_hash}``. + + A skill tree is any tree containing a ``SKILL.md`` blob entry. We walk the + root tree; a subtree with a SKILL.md is treated as a skill leaf keyed by + its path, so category nesting is preserved. + """ + result: Dict[str, str] = {} + + def _walk(tree_hash: str, prefix: str) -> None: + tree = client.get_tree_json(tree_hash) + entries = tree.get("entries", []) + has_skill_md = any( + e.get("name") == "SKILL.md" and e.get("kind") == KIND_BLOB for e in entries + ) + if has_skill_md and prefix: + result[prefix] = tree_hash + return + for e in entries: + if e.get("kind") == KIND_TREE: + child_prefix = f"{prefix}/{e['name']}" if prefix else e["name"] + _walk(e["hash"], child_prefix) + + _walk(root_tree_hash, "") + return result + + +def _check_version(caps: Dict[str, Any]) -> None: + """Reject an incompatible server major version (contract §1).""" + ver = str(caps.get("hsp_version") or "") + major = ver.split(".", 1)[0] + if major != HSP_VERSION: + raise HSPError(f"incompatible HSP version {ver!r} (client speaks {HSP_VERSION})") + + +# --------------------------------------------------------------------------- +# Push +# --------------------------------------------------------------------------- + +def push_skills( + client: Optional["HSPClient"] = None, + *, + skill_names: Optional[List[str]] = None, + identity: Optional[Dict[str, Any]] = None, + message: str = "hermes skill sync", +) -> Dict[str, Any]: + """Push opted-in skills to the owner's HEAD (contract §4). + + Uploads all new objects, then CAS-es ``refs/user//HEAD``. On a 409, + fetches the actual head, three-way merges, and retries once (§4.4 / M1-C). + Returns a result dict; never raises for the inert / no-op cases. + """ + if identity is None: + identity = resolve_identity() + owner = identity["owner"] + if client is None: + base = resolve_sync_base_url() + if not base: + return {"ok": False, "reason": "no sync base url configured", "noop": True} + client = HSPClient(base, identity["api_key"]) + + if skill_names is None: + skill_names = list_synced_skill_names() + if not skill_names: + return {"ok": True, "reason": "no skills opted into sync", "noop": True} + + caps = client.capabilities() + _check_version(caps) + max_bytes = int(caps.get("max_object_bytes") or DEFAULT_MAX_OBJECT_BYTES) + + objects, root_hash, _ = snapshot_profile(skill_names, max_object_bytes=max_bytes) + + manifest = read_sync_manifest() + base_head = manifest.get("head") + + # Idempotency: if the profile-root tree is unchanged since our last push, + # there is nothing to propagate -- skip building an empty commit (contract + # objects are immutable, so an identical tree hash means identical content). + if base_head and manifest.get("root") == root_hash: + return {"ok": True, "head": base_head, "reason": "unchanged", "noop": True} + + device = stable_device_id() + parents = [base_head] if base_head else [] + commit_hash = build_commit( + root_hash, parents, owner=owner, device=device, message=message, objects=objects + ) + + client.put_objects(objects.objects) + ref = user_head_ref(owner) + + try: + client.cas_ref(ref, base_head, commit_hash) + manifest["head"] = commit_hash + manifest["root"] = root_hash + write_sync_manifest(manifest) + return {"ok": True, "head": commit_hash, "pushed_objects": len(objects)} + except HSPConflict as conflict: + return _resolve_push_conflict( + client, identity, conflict.actual, root_hash, commit_hash, + objects, skill_names, message, base_head, + ) + + +# --------------------------------------------------------------------------- +# Conflict resolution / three-way merge (contract §4.4, M1-C) +# +# On a 409 the server hands back the actual head. We fetch it, three-way merge +# per skill against the base we forked from, reusing the origin/user/incoming +# decision semantics of skills_sync.py (_is_tracked_user_modification + +# the decision block at skills_sync.py:619-643): +# +# * base == ours == theirs -> nothing to do +# * ours == base, theirs moved -> take theirs (fast-forward incoming) +# * theirs == base, ours moved -> keep ours (our local edit) +# * both moved, ours == theirs -> converged; take either +# * both moved, differ -> TRUE OVERLAP -> conflict head +# +# Non-overlapping merges (each side changed a DIFFERENT skill) produce a merge +# commit (2 parents) and retry the CAS. A true overlap (both sides changed the +# SAME skill differently) is written to refs/user//conflict/ and +# surfaced for out-of-band resolution. +# --------------------------------------------------------------------------- + +def _resolve_push_conflict( + client: "HSPClient", + identity: Dict[str, Any], + actual_head: str, + our_root: str, + our_commit: str, + objects: "ObjectSet", + skill_names: List[str], + message: str, + base_head: Optional[str], +) -> Dict[str, Any]: + owner = identity["owner"] + device = stable_device_id() + + theirs_root = _root_tree_of_commit(client, actual_head) + base_root = _root_tree_of_commit(client, base_head) if base_head else None + + ours_trees = _skill_trees_of_root(client, our_root) + theirs_trees = _skill_trees_of_root(client, theirs_root) + base_trees = _skill_trees_of_root(client, base_root) if base_root else {} + + merged: Dict[str, str] = {} + overlaps: List[str] = [] + all_paths = set(ours_trees) | set(theirs_trees) | set(base_trees) + for path in all_paths: + o = ours_trees.get(path) + t = theirs_trees.get(path) + b = base_trees.get(path) + decision = _merge_skill(b, o, t) + if decision == "overlap": + overlaps.append(path) + # Keep OURS on the surfaced conflict head; theirs is retained + # server-side under the conflict ref for out-of-band resolution. + if o is not None: + merged[path] = o + elif decision == "ours" and o is not None: + merged[path] = o + elif decision == "theirs" and t is not None: + merged[path] = t + elif decision == "either": + merged[path] = o if o is not None else t # type: ignore[assignment] + # decision == "none": skill deleted on the winning side -> drop + + if overlaps: + # TRUE OVERLAP -> write a conflict head and surface it (M1-C). + n = _next_conflict_index(client, owner) + conflict_ref = user_conflict_ref(owner, n) + try: + client.cas_ref(conflict_ref, None, our_commit) + except HSPConflict: + pass # someone else grabbed this index; the head still exists + return { + "ok": False, + "conflict": True, + "conflict_ref": conflict_ref, + "overlapping_skills": sorted(overlaps), + "actual_head": actual_head, + "message": ( + f"{len(overlaps)} skill(s) changed on both sides; wrote " + f"{conflict_ref}. Resolve out-of-band (hermes sync / NAS UI)." + ), + } + + # Non-overlap -> build a merge commit (parents: base->actual, ours) and + # retry the CAS against the actual head. + merge_objects = ObjectSet() + # Re-add our objects so the merge push is self-contained (idempotent). + for h, (kind, data) in objects.objects.items(): + merge_objects.objects[h] = (kind, data) + merged_root = _assemble_root_from_skill_trees(client, merged, merge_objects) + merge_commit = build_commit( + merged_root, + [actual_head, our_commit], + owner=owner, + device=device, + message=f"merge: {message}", + objects=merge_objects, + ) + client.put_objects(merge_objects.objects) + try: + client.cas_ref(user_head_ref(owner), actual_head, merge_commit) + except HSPConflict as c2: + return { + "ok": False, + "conflict": True, + "message": f"merge CAS lost again (head now {c2.actual}); retry sync.", + "actual_head": c2.actual, + } + manifest = read_sync_manifest() + manifest["head"] = merge_commit + manifest["root"] = merged_root + write_sync_manifest(manifest) + return {"ok": True, "head": merge_commit, "merged": True} + + +def _merge_skill(base: Optional[str], ours: Optional[str], theirs: Optional[str]) -> str: + """Three-way decision for one skill's tree hash. + + Returns one of: ``ours``, ``theirs``, ``either``, ``overlap``, ``none``. + Mirrors the origin/user/incoming decision block of skills_sync.py:619-643: + a side "modified" the skill when its hash differs from the common base + (analogous to ``_is_tracked_user_modification(origin, current)``). + """ + if ours == theirs: + return "either" if ours is not None else "none" + ours_changed = ours != base + theirs_changed = theirs != base + if ours_changed and not theirs_changed: + return "ours" + if theirs_changed and not ours_changed: + return "theirs" + # both changed and differ + return "overlap" + + +def _assemble_root_from_skill_trees( + client: "HSPClient", skill_trees: Dict[str, str], objects: "ObjectSet" +) -> str: + """Build a profile-root tree object from ``{posix_rel_path: tree_hash}``. + + Rebuilds the intermediate category trees. The referenced skill trees are + assumed already durable (they came from either side of the merge); only + the new intermediate/root tree objects are added to *objects*. + """ + root: Dict[str, Any] = {} + for path, tree_hash in skill_trees.items(): + parts = PurePosixPath(path).parts + node = root + for part in parts[:-1]: + node = node.setdefault(part, {}) + node[parts[-1]] = {"__tree__": tree_hash} + return _build_root_tree(root, objects) + + +def _next_conflict_index(client: "HSPClient", owner: str) -> int: + """Pick the next free conflict ref index for the owner.""" + try: + refs = client.get_refs(f"refs/user/{owner}/conflict/") + except HSPError: + return 1 + used = [] + for r in refs: + name = r.get("name", "") + tail = name.rsplit("/", 1)[-1] + if tail.isdigit(): + used.append(int(tail)) + return (max(used) + 1) if used else 1 + + +# --------------------------------------------------------------------------- +# Pull +# --------------------------------------------------------------------------- + +def pull_skills( + client: Optional["HSPClient"] = None, + *, + identity: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Pull the owner's HEAD and materialize opted-in skills to disk. + + Fetches ``refs/user//HEAD``; if it advanced past our recorded head, + walks the profile-root tree and writes each skill tree into + ~/.hermes/skills/. Only paths the user has opted into (``sync: true``) are + materialized, so a pull never resurrects a skill the user hasn't chosen. + Best-effort; returns a result dict. + """ + if identity is None: + identity = resolve_identity() + owner = identity["owner"] + if client is None: + base = resolve_sync_base_url() + if not base: + return {"ok": False, "reason": "no sync base url configured", "noop": True} + client = HSPClient(base, identity["api_key"]) + + caps = client.capabilities() + _check_version(caps) + + refs = client.get_refs(user_head_ref(owner)) + head = None + for r in refs: + if r.get("name") == user_head_ref(owner): + head = r.get("hash") + break + if not head: + return {"ok": True, "reason": "no remote HEAD yet", "noop": True} + + manifest = read_sync_manifest() + if head == manifest.get("head"): + return {"ok": True, "reason": "already up to date", "head": head, "noop": True} + + root_tree = _root_tree_of_commit(client, head) + remote_trees = _skill_trees_of_root(client, root_tree) + + opted_in = set(_opted_in_rel_paths()) + updated = [] + for path, tree_hash in remote_trees.items(): + # Opt-in gate on pull: only materialize skills the user chose to sync. + if opted_in and path not in opted_in: + continue + dest = _skills_dir() / path + materialize_tree(client, tree_hash, dest) + updated.append(path) + + manifest["head"] = head + write_sync_manifest(manifest) + return {"ok": True, "head": head, "updated": sorted(updated)} + + +def _opted_in_rel_paths() -> List[str]: + """Relative posix paths of skills the user has opted into sync.""" + paths = [] + for name in list_synced_skill_names(): + rel = _skill_rel_path(name) + if rel is not None: + paths.append(rel.as_posix()) + return paths + + +# --------------------------------------------------------------------------- +# Gated public entrypoints (gate-and-swallow) +# +# maybe_pull_skills / maybe_push_skills clone the shape of the curator's +# maybe_run_curator (agent/curator.py:1998): best-effort, never raise, return +# a result dict or None. The DEV-PHASE gate is checked first -- sync is inert +# (no push, no pull, no-op) unless tool_gateway_admin === true on the token. +# --------------------------------------------------------------------------- + +def maybe_push_skills(*, message: str = "hermes skill sync") -> Optional[Dict[str, Any]]: + """Best-effort push if all gates pass. Returns a result dict or None. + Never raises. Called from the debounced skill_manage push hook.""" + try: + identity = resolve_identity() + if not identity.get("dev_gate_ok"): + return None # DEV-PHASE gate: inert without tool_gateway_admin + if not resolve_sync_base_url(): + return None + if not list_synced_skill_names(): + return None + return push_skills(identity=identity, message=message) + except Exception as e: + logger.debug("skills_sync_client: maybe_push_skills failed: %s", e, exc_info=True) + return None + + +def maybe_pull_skills() -> Optional[Dict[str, Any]]: + """Best-effort pull if all gates pass. Returns a result dict or None. + Never raises. Invoked at the curator tick sites (gateway housekeeping loop + + CLI startup).""" + try: + identity = resolve_identity() + if not identity.get("dev_gate_ok"): + return None # DEV-PHASE gate: inert without tool_gateway_admin + if not resolve_sync_base_url(): + return None + return pull_skills(identity=identity) + except Exception as e: + logger.debug("skills_sync_client: maybe_pull_skills failed: %s", e, exc_info=True) + return None + + +def sync_status() -> Dict[str, Any]: + """Return a status snapshot for ``hermes sync status``. Never raises.""" + status: Dict[str, Any] = { + "dev_gate_ok": False, + "logged_in": False, + "base_url": resolve_sync_base_url(), + "opted_in_skills": [], + "local_head": None, + "owner": None, + } + try: + identity = resolve_identity() + status["logged_in"] = True + status["owner"] = identity.get("owner") + status["dev_gate_ok"] = bool(identity.get("dev_gate_ok")) + except SyncInertError: + pass + except Exception as e: + logger.debug("skills_sync_client: sync_status identity failed: %s", e) + try: + status["opted_in_skills"] = list_synced_skill_names() + status["local_head"] = read_sync_manifest().get("head") + except Exception: + pass + return status