diff --git a/api/skill_sync.py b/api/skill_sync.py index 666c083..7caca87 100644 --- a/api/skill_sync.py +++ b/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, diff --git a/openspec/changes/skill-management-console/tasks.md b/openspec/changes/skill-management-console/tasks.md index f25cea8..f7b89c6 100644 --- a/openspec/changes/skill-management-console/tasks.md +++ b/openspec/changes/skill-management-console/tasks.md @@ -38,9 +38,9 @@ ## 7. Sync repoint + runner wiring + inventory report -- [ ] 7.1 Repoint `api/skill_sync.py`'s concrete client at the Cloud API per-host sync endpoint (host-scoped bearer; `since_version` incremental; full-replace on first/stale); keep the `SubscriptionClient` Protocol / `SyncDelta` shape. The agent's "subscription_id" becomes its host identifier. +- [x] 7.1 Repoint `api/skill_sync.py`'s concrete client at the Cloud API per-host sync endpoint (host-scoped bearer; `since_version` incremental; full-replace on first/stale); keep the `SubscriptionClient` Protocol / `SyncDelta` shape. The agent's "subscription_id" becomes its host identifier. - [ ] 7.2 Construct and start `SkillSyncRunner` in the host-agent app bootstrap behind the existing cloud-transport configuration; configurable poll interval; failures non-fatal (preserve cache + record error). -- [ ] 7.3 On a sync `removed_ids` entry that has a local override, trigger the fork (D9) via `LocalSkillStore.fork_override_to_local`. +- [x] 7.3 On a sync `removed_ids` entry that has a local override, trigger the fork (D9) via `LocalSkillStore.fork_override_to_local`. - [ ] 7.4 Add a periodic best-effort local-skill inventory report from the agent to the Cloud inventory endpoint (metadata only). - [ ] 7.5 Tests: incremental apply, full-replace, fork-on-revocation end-to-end, runner lifecycle, inventory report payload + failure-isolation. diff --git a/storage/skill_catalog.py b/storage/skill_catalog.py index e6bafa0..31ef1b8 100644 --- a/storage/skill_catalog.py +++ b/storage/skill_catalog.py @@ -143,6 +143,18 @@ class SkillCatalogStore: (subscription_id, 1 if active else 0), ) + def _get_subscription_version(self, subscription_id: str) -> int | None: + """Last successfully applied ``latest_version`` for a subscription.""" + with self._connect() as conn: + row = conn.execute( + "SELECT last_synced_version FROM subscriptions WHERE id = ?", + (subscription_id,), + ).fetchone() + if row is None: + return None + value = row["last_synced_version"] + return int(value) if value is not None else None + def _set_subscription_state( self, subscription_id: str, diff --git a/tests/test_cloud_skill_sync.py b/tests/test_cloud_skill_sync.py new file mode 100644 index 0000000..6869956 --- /dev/null +++ b/tests/test_cloud_skill_sync.py @@ -0,0 +1,137 @@ +"""Tests for the Cloud API skill sync client and runner enhancements: +incremental since_version forwarding and fork-on-revocation (design D3/D9). +""" +from __future__ import annotations + +import httpx +import pytest + +from api.skill_sync import ( + CloudApiSkillClient, + SkillSyncRunner, + SyncDelta, +) +from skills_learning.models import KnowledgeSkill, SkillMetadata +from storage.local_skills import LocalSkillStore +from storage.skill_catalog import SkillCatalogStore + + +def _cloud_skill_dict(skill_id: str, name: str, content: str) -> dict: + return { + "id": skill_id, + "name": name, + "kind": "knowledge", + "description": "d", + "tags": [], + "revision": 3, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": "2026-01-01T00:00:00+00:00", + "content": content, + "steps": [], + "parameters": {}, + } + + +def test_cloud_api_skill_client_parses_revision_to_version(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["since"] = request.url.params.get("since_version") + captured["auth"] = request.headers.get("authorization") + return httpx.Response( + 200, + json={ + "skills": [_cloud_skill_dict("c1", "Cloud One", "body")], + "removed_ids": [], + "latest_version": 7, + "is_full_replace": False, + }, + ) + + transport = httpx.MockTransport(handler) + client = CloudApiSkillClient( + "https://cloud.example/", + host_token="host-token", + client=httpx.Client(transport=transport), + ) + delta = client.fetch_entitled_skills("host-1", since_version=5) + assert captured["since"] == "5" + assert captured["auth"] == "Bearer host-token" + assert delta.latest_version == 7 + assert delta.is_full_replace is False + [skill] = delta.skills + assert skill.id == "c1" + assert skill.metadata.version == 3 # revision mapped to version + assert isinstance(skill, KnowledgeSkill) + assert skill.content == "body" + + +class _RecordingClient: + def __init__(self, deltas: list[SyncDelta]) -> None: + self._deltas = list(deltas) + self.seen_since: list[int | None] = [] + + def fetch_entitled_skills(self, subscription_id, since_version=None): + self.seen_since.append(since_version) + return self._deltas.pop(0) + + +def test_runner_forwards_since_version_after_first_full_replace(tmp_path): + store = SkillCatalogStore(db_path=tmp_path / "skills.sqlite3") + deltas = [ + SyncDelta( + skills=[KnowledgeSkill(metadata=SkillMetadata(id="c1", name="A"), content="x")], + removed_ids=[], + latest_version=1, + is_full_replace=True, + ), + SyncDelta( + skills=[], + removed_ids=[], + latest_version=1, + is_full_replace=False, + ), + ] + client = _RecordingClient(deltas) + runner = SkillSyncRunner( + store=store, client=client, subscriptions=["host-1"], poll_interval=999.0 + ) + runner.tick() + assert client.seen_since[0] is None # first sync is full + runner.tick() + assert client.seen_since[1] == 1 # second forwards last version + + +def test_runner_forks_override_on_revocation(tmp_path): + store = SkillCatalogStore(db_path=tmp_path / "skills.sqlite3") + local = LocalSkillStore(db_path=tmp_path / "local.sqlite3") + + # Seed the synced store with a cloud skill and a local override for it. + store._register_subscription("host-1") + cloud_skill = KnowledgeSkill( + metadata=SkillMetadata(id="c-cloud", name="Cloud", source="cloud"), + content="original", + ) + store._apply_sync_upsert(cloud_skill, "host-1") + local.upsert_override("c-cloud", KnowledgeSkill( + metadata=SkillMetadata(id="c-cloud", name="Cloud"), + content="my override", + )) + store._set_subscription_state("host-1", last_synced_version=1) + + client = _RecordingClient([ + SyncDelta(skills=[], removed_ids=["c-cloud"], latest_version=2, is_full_replace=False) + ]) + runner = SkillSyncRunner( + store=store, client=client, subscriptions=["host-1"], + poll_interval=999.0, local_store=local, + ) + runner.tick() + + # Override forked into a standalone local skill. + overrides_after = local.list_overrides() + assert overrides_after == [] + forked = [s for s in local.list_local() if s.name == "Cloud"] + assert len(forked) == 1 + # Cloud skill removed from the synced store. + assert store.get_skill("c-cloud", {"host-1"}) is None