feat(skill-catalog-subscription): synced catalog + HTTP sync + MCP tools
Consumes the external Subscription Platform as source of truth for skill content; reuses skills_learning domain models (extended with KnowledgeSkill) and workflow.skill_exec resolver. HTTP/MCP deps land in api/ per CONSTITUTION.md; synced skills use a physically separate SQLite file (tasks/skills.sqlite3) to preserve the skill-authoring capability boundary. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,277 @@
|
||||
"""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.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 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,
|
||||
) -> None:
|
||||
self.store = store
|
||||
self.client = client
|
||||
self.subscriptions = list(subscriptions)
|
||||
self.poll_interval = poll_interval
|
||||
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:
|
||||
try:
|
||||
delta = self.client.fetch_entitled_skills(subscription_id)
|
||||
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:
|
||||
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
|
||||
Reference in New Issue
Block a user