feat(skills): cloud sync client + incremental sync + fork-on-revocation

Adds CloudApiSkillClient (Cloud API per-host sync endpoint + inventory
report), forwards since_version for incremental sync (full-replace on
first/stale), and forks a local override into a standalone local skill
when its cloud skill is revoked (design D9). Skill-side tests green (87
passed).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

api/skill_sync.py

@@ -21,6 +21,7 @@ from skills_learning.models import (
     KnowledgeSkill,
     Skill,
 )
+from storage.local_skills import LocalSkillStore
 from storage.skill_catalog import SkillCatalogStore
 
 log = logging.getLogger(__name__)
@@ -144,6 +145,100 @@ def _parse_sync_payload(payload: dict[str, Any]) -> SyncDelta:
     )
 
 
+class CloudApiSkillClient:
+    """Concrete client for this project's Cloud API per-host skill sync endpoint.
+
+    The ``subscription_id`` passed to :meth:`fetch_entitled_skills` is the
+    agent's host identifier; the endpoint is
+    ``GET /internal/v1/hosts/{host_id}/skills/sync`` authenticated with the
+    same host-scoped bearer used for heartbeat/planner-decision.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        host_token: str,
+        *,
+        timeout: float = 30.0,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.host_token = host_token
+        self.timeout = timeout
+        self._client = client
+
+    def fetch_entitled_skills(
+        self,
+        subscription_id: str,
+        since_version: int | None = None,
+    ) -> SyncDelta:
+        url = f"{self.base_url}/internal/v1/hosts/{subscription_id}/skills/sync"
+        params: dict[str, Any] = {}
+        if since_version is not None:
+            params["since_version"] = str(since_version)
+        response = self._ensure_client().get(
+            url,
+            params=params or None,
+            headers={"Authorization": f"Bearer {self.host_token}"},
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return _parse_cloud_sync_payload(response.json())
+
+    def report_inventory(
+        self, host_id: str, inventory: list[dict[str, Any]]
+    ) -> None:
+        """Best-effort local-skill inventory report to the Cloud (design D7)."""
+        response = self._ensure_client().post(
+            f"{self.base_url}/internal/v1/hosts/{host_id}/skills/inventory",
+            json={"skills": inventory},
+            headers={"Authorization": f"Bearer {self.host_token}"},
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+
+    def close(self) -> None:
+        if self._client is not None:
+            self._client.close()
+            self._client = None
+
+    def _ensure_client(self) -> httpx.Client:
+        if self._client is None:
+            self._client = httpx.Client()
+        return self._client
+
+
+def _parse_cloud_sync_payload(payload: dict[str, Any]) -> SyncDelta:
+    """Parse the Cloud API sync response into a SyncDelta.
+
+    The Cloud skill payloads carry ``revision`` (mapped to the local
+    ``version``) and kind-specific ``content``/``steps``/``parameters`` fields
+    that line up with :class:`skills_learning.models` ``from_dict``.
+    """
+    skills: list[Skill] = []
+    for item in payload.get("skills") or []:
+        normalized = dict(item)
+        if "version" not in normalized and "revision" in normalized:
+            normalized["version"] = normalized["revision"]
+        source = "cloud"
+        kind = normalized.get("kind", "knowledge")
+        normalized["source"] = source
+        if kind == "flow_template":
+            skills.append(FlowTemplateSkill.from_dict(normalized))
+        else:
+            skills.append(KnowledgeSkill.from_dict(normalized))
+    removed_ids = [str(rid) for rid in payload.get("removed_ids") or []]
+    latest_raw = payload.get("latest_version")
+    latest_version = int(latest_raw) if latest_raw is not None else None
+    is_full_replace = bool(payload.get("is_full_replace", True))
+    return SyncDelta(
+        skills=skills,
+        removed_ids=removed_ids,
+        latest_version=latest_version,
+        is_full_replace=is_full_replace,
+    )
+
+
 class SkillSyncRunner:
     """Drives periodic sync between the Subscription Platform and local catalog.
 
@@ -160,11 +255,13 @@ class SkillSyncRunner:
         client: SubscriptionClient,
         subscriptions: list[str],
         poll_interval: float = 300.0,
+        local_store: LocalSkillStore | None = None,
     ) -> None:
         self.store = store
         self.client = client
         self.subscriptions = list(subscriptions)
         self.poll_interval = poll_interval
+        self.local_store = local_store
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
         self._tick_lock = threading.Lock()
@@ -209,8 +306,11 @@ class SkillSyncRunner:
             self._stop.wait(self.poll_interval)
 
     def _sync_one(self, subscription_id: str) -> SyncOutcome:
+        since_version = self.store._get_subscription_version(subscription_id)
         try:
-            delta = self.client.fetch_entitled_skills(subscription_id)
+            delta = self.client.fetch_entitled_skills(
+                subscription_id, since_version=since_version
+            )
         except Exception as exc:
             log.warning(
                 "skill sync fetch failed for %s: %s", subscription_id, exc
@@ -233,6 +333,11 @@ class SkillSyncRunner:
             for skill in delta.skills:
                 self.store._apply_sync_upsert(skill, subscription_id)
             for skill_id in delta.removed_ids:
+                # Fork-on-revocation (design D9): if a local override shadows
+                # this cloud skill, promote it to a standalone local skill
+                # before the cloud id disappears from the synced store.
+                if self.local_store is not None:
+                    self.local_store.fork_override_to_local(skill_id)
                 self.store._apply_sync_remove(skill_id)
             self.store._set_subscription_state(
                 subscription_id,