q792602257andClaude Opus 4.6
a8ba2312fc
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>
383 lines
13 KiB
Python
383 lines
13 KiB
Python
"""Skill catalog sync with the external Subscription Platform.
|
|
|
|
API layer per CONSTITUTION.md: HTTP dependencies (``httpx``) live here, and
|
|
this module is the sole caller of :mod:`storage.skill_catalog`'s private
|
|
``_apply_sync_*`` methods — the "Subscription Platform is sole source of
|
|
truth" contract is enforced by that import boundary.
|
|
|
|
Push (webhook) is optional; the baseline pull loop is correct standalone.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Protocol
|
|
|
|
import httpx
|
|
|
|
from skills_learning.models import (
|
|
FlowTemplateSkill,
|
|
KnowledgeSkill,
|
|
Skill,
|
|
)
|
|
from storage.local_skills import LocalSkillStore
|
|
from storage.skill_catalog import SkillCatalogStore
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class SubscriptionClient(Protocol):
|
|
"""Contract for talking to the Subscription Platform.
|
|
|
|
Concrete implementations may use HTTP, an in-process mock, or any other
|
|
transport — :class:`SkillSyncRunner` only depends on this Protocol, so
|
|
the real Subscription Platform API can be wired in by adjusting only the
|
|
concrete client class (see design.md Open Questions).
|
|
"""
|
|
|
|
def fetch_entitled_skills(
|
|
self,
|
|
subscription_id: str,
|
|
since_version: int | None = None,
|
|
) -> "SyncDelta":
|
|
"""Return the skills entitled to ``subscription_id``.
|
|
|
|
If ``since_version`` is given, the implementation MAY return only
|
|
skills changed since that version (incremental). If omitted or if
|
|
the implementation does not support incremental fetch, returns the
|
|
full entitled set (the runner treats ``is_full_replace`` as truthy).
|
|
"""
|
|
...
|
|
|
|
|
|
@dataclass
|
|
class SyncDelta:
|
|
"""Result of a single ``fetch_entitled_skills`` call."""
|
|
|
|
skills: list[Skill]
|
|
removed_ids: list[str] = field(default_factory=list)
|
|
latest_version: int | None = None
|
|
is_full_replace: bool = True
|
|
|
|
|
|
@dataclass
|
|
class SyncOutcome:
|
|
"""Per-subscription result of one ``SkillSyncRunner.tick()`` pass."""
|
|
|
|
success: bool
|
|
error: str | None
|
|
fetched: int
|
|
|
|
|
|
class HttpSubscriptionClient:
|
|
"""Concrete HTTP implementation of :class:`SubscriptionClient`.
|
|
|
|
The request/response shape is shaped to an assumed Subscription Platform
|
|
REST API; once the real API is finalized, only this class needs to
|
|
change (see design.md D3 and Open Questions).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
auth_token: str | None = None,
|
|
*,
|
|
timeout: float = 30.0,
|
|
client: httpx.Client | None = None,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.auth_token = auth_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}/subscriptions/{subscription_id}/skills"
|
|
params: dict[str, Any] = {}
|
|
if since_version is not None:
|
|
params["since_version"] = str(since_version)
|
|
headers: dict[str, str] = {}
|
|
if self.auth_token:
|
|
headers["Authorization"] = f"Bearer {self.auth_token}"
|
|
|
|
response = self._ensure_client().get(
|
|
url,
|
|
params=params or None,
|
|
headers=headers or None,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return _parse_sync_payload(response.json())
|
|
|
|
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_sync_payload(payload: dict[str, Any]) -> SyncDelta:
|
|
"""Parse the assumed Subscription Platform response into a SyncDelta."""
|
|
skills: list[Skill] = []
|
|
for item in payload.get("skills") or []:
|
|
kind = item.get("kind", "knowledge")
|
|
if kind == "flow_template":
|
|
skills.append(FlowTemplateSkill.from_dict(item))
|
|
else:
|
|
skills.append(KnowledgeSkill.from_dict(item))
|
|
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 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.
|
|
|
|
Push (webhook) is optional: poll-only operation is the baseline. The
|
|
runner does not auto-start any thread on import — call
|
|
:meth:`start_background` to begin polling, or :meth:`tick` for manual
|
|
control (used by tests and by the webhook trigger).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
store: SkillCatalogStore,
|
|
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()
|
|
for sub_id in self.subscriptions:
|
|
self.store._register_subscription(sub_id)
|
|
|
|
def tick(self) -> dict[str, SyncOutcome]:
|
|
"""Run one sync pass across all configured subscriptions.
|
|
|
|
Failures do not raise; the cache is preserved and the failure is
|
|
recorded in the subscriptions row (``last_error`` /
|
|
``last_synced_at``) for observability.
|
|
"""
|
|
with self._tick_lock:
|
|
return {sub_id: self._sync_one(sub_id) for sub_id in self.subscriptions}
|
|
|
|
def sync_one(self, subscription_id: str) -> SyncOutcome:
|
|
"""Sync a single subscription immediately (used by the webhook)."""
|
|
with self._tick_lock:
|
|
return self._sync_one(subscription_id)
|
|
|
|
def start_background(self) -> None:
|
|
"""Start a daemon thread that calls :meth:`tick` on the configured interval."""
|
|
if self._thread is not None:
|
|
return
|
|
self._stop.clear()
|
|
self._thread = threading.Thread(target=self._run_forever, daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop_background(self) -> None:
|
|
self._stop.set()
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=5.0)
|
|
self._thread = None
|
|
|
|
def _run_forever(self) -> None:
|
|
while not self._stop.is_set():
|
|
try:
|
|
self.tick()
|
|
except Exception:
|
|
log.exception("skill sync tick raised unexpectedly")
|
|
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, since_version=since_version
|
|
)
|
|
except Exception as exc:
|
|
log.warning(
|
|
"skill sync fetch failed for %s: %s", subscription_id, exc
|
|
)
|
|
self.store._set_subscription_state(
|
|
subscription_id,
|
|
last_error=f"{type(exc).__name__}: {exc}",
|
|
bump_synced_at=True,
|
|
)
|
|
return SyncOutcome(success=False, error=str(exc), fetched=0)
|
|
|
|
if delta.is_full_replace:
|
|
self.store._apply_sync_replace_all(
|
|
subscription_id,
|
|
delta.skills,
|
|
latest_version=delta.latest_version,
|
|
last_error=None,
|
|
)
|
|
else:
|
|
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,
|
|
last_synced_version=delta.latest_version,
|
|
last_error=None,
|
|
)
|
|
return SyncOutcome(success=True, error=None, fetched=len(delta.skills))
|
|
|
|
|
|
def register_skill_sync_webhook(app: Any, runner: SkillSyncRunner) -> Any:
|
|
"""Attach ``POST /webhooks/skill-sync`` to a FastAPI app.
|
|
|
|
On notification, triggers an immediate ``runner.tick()`` out-of-cycle.
|
|
Optional — the poll loop remains correct standalone. The endpoint
|
|
accepts an optional JSON body ``{"subscription_id": "..."}`` to sync
|
|
only that subscription; if omitted, syncs all configured subscriptions.
|
|
Returns the runner so the caller can chain.
|
|
"""
|
|
from fastapi import Body, FastAPI
|
|
|
|
if not isinstance(app, FastAPI):
|
|
raise TypeError("register_skill_sync_webhook requires a FastAPI app")
|
|
|
|
@app.post("/webhooks/skill-sync")
|
|
async def _skill_sync_webhook(
|
|
body: dict | None = Body(default=None),
|
|
) -> dict[str, Any]:
|
|
sub_id = (body or {}).get("subscription_id")
|
|
if sub_id:
|
|
outcome = runner.sync_one(str(sub_id))
|
|
outcomes = {str(sub_id): outcome}
|
|
else:
|
|
outcomes = runner.tick()
|
|
return {
|
|
"ok": True,
|
|
"outcomes": {
|
|
sid: {"success": o.success, "error": o.error, "fetched": o.fetched}
|
|
for sid, o in outcomes.items()
|
|
},
|
|
}
|
|
|
|
return runner
|