feat(cloud): cloud-managed skill store + per-host entitlement + sync versioning

Adds cloud/skills.py (domain + service), SQLAlchemy models and Alembic
migration 0010_skill_management (cloud_skills, cloud_skill_entitlements,
cloud_skill_sync_state, a per-host changelog, and host_skill_inventory),
and repository methods with a monotonic per-host entitlement_version that
drives correct incremental fetch_host_delta. cloud-api suite green (41
passed); HEAD_REVISION bumped to 0010.

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

apps/cloud-api/tests/test_cloud_skill_management.py

@@ -0,0 +1,143 @@
+"""Repository + service tests for Cloud-managed skills and per-host sync.
+
+Uses an in-memory SQLite engine. Covers skill CRUD, per-host entitlement, the
+monotonic entitlement_version bump, and incremental vs full-replace
+fetch_host_delta semantics (design D2/D3).
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.pool import StaticPool
+
+from cloud.db_models import Base
+from cloud.skills import (
+    CloudSkillConflictError,
+    CloudSkillValidationError,
+    CloudSkillService,
+)
+from cloud.sql_repository import SQLAlchemyCloudRepository
+from core.models import utc_now
+
+
+@pytest.fixture
+def service():
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    repo = SQLAlchemyCloudRepository(engine=engine, create_schema=False)
+    return CloudSkillService(repo)
+
+
+def _knowledge_payload(name: str, content: str = "body") -> dict:
+    return dict(
+        name=name,
+        kind="knowledge",
+        description="d",
+        tags=["t"],
+        content=content,
+        steps_json="[]",
+        parameters_json="{}",
+    )
+
+
+def test_create_list_get_skill(service):
+    now = utc_now()
+    created = service.create_skill(now=now, **_knowledge_payload("Alpha"))
+    assert created.name == "Alpha"
+    assert created.kind == "knowledge"
+    [fetched] = service.list_skills()
+    assert fetched.id == created.id
+    assert service.get_skill(created.id).content == "body"
+
+
+def test_create_rejects_duplicate_name(service):
+    service.create_skill(now=utc_now(), **_knowledge_payload("Alpha"))
+    with pytest.raises(CloudSkillConflictError):
+        service.create_skill(now=utc_now(), **_knowledge_payload("Alpha"))
+
+
+def test_create_rejects_blank_content_for_knowledge(service):
+    payload = _knowledge_payload("Alpha", content="   ")
+    with pytest.raises(CloudSkillValidationError):
+        service.create_skill(now=utc_now(), **payload)
+
+
+def test_grant_revoke_entitlement_drives_delta(service):
+    now = utc_now()
+    skill = service.create_skill(now=now, **_knowledge_payload("Alpha"))
+    host = "host-1"
+
+    # First sync: no entitlements yet -> empty full replace.
+    delta = service.fetch_host_delta(host, since_version=None)
+    assert delta.is_full_replace is True
+    assert delta.skills == []
+    assert delta.latest_version == 0
+
+    # Grant -> version bumps, next full sync sees the skill.
+    service.grant_entitlement(skill.id, host, now=now)
+    full = service.fetch_host_delta(host, since_version=None)
+    assert [s.id for s in full.skills] == [skill.id]
+    assert full.latest_version == 1
+
+    # Incremental from 0 returns the grant.
+    incr = service.fetch_host_delta(host, since_version=0)
+    assert incr.is_full_replace is False
+    assert [s.id for s in incr.skills] == [skill.id]
+    assert incr.removed_ids == []
+
+    # Revoke -> version bumps, incremental reports removal.
+    service.revoke_entitlement(skill.id, host, now=now)
+    after = service.fetch_host_delta(host, since_version=incr.latest_version)
+    assert after.removed_ids == [skill.id]
+    assert after.skills == []
+
+
+def test_skill_content_update_notifies_entitled_hosts(service):
+    now = utc_now()
+    skill = service.create_skill(now=now, **_knowledge_payload("Alpha", "v1"))
+    service.grant_entitlement(skill.id, "host-1", now=now)
+    baseline = service.fetch_host_delta("host-1", since_version=None).latest_version
+
+    updated = service.update_skill(
+        skill.id, now=utc_now(), **_knowledge_payload("Alpha", "v2")
+    )
+    assert updated.revision == 2
+    incr = service.fetch_host_delta("host-1", since_version=baseline)
+    assert [s.id for s in incr.skills] == [skill.id]
+    assert incr.skills[0].content == "v2"
+
+
+def test_delete_skill_removes_and_notifies_entitled_hosts(service):
+    now = utc_now()
+    skill = service.create_skill(now=now, **_knowledge_payload("Alpha"))
+    service.grant_entitlement(skill.id, "host-1", now=now)
+    baseline = service.fetch_host_delta("host-1", since_version=None).latest_version
+
+    service.delete_skill(skill.id)
+    assert service.get_skill(skill.id) is None
+    after = service.fetch_host_delta("host-1", since_version=baseline)
+    assert after.removed_ids == [skill.id]
+
+
+def test_stale_since_version_falls_back_to_full_replace(service):
+    now = utc_now()
+    skill = service.create_skill(now=now, **_knowledge_payload("Alpha"))
+    service.grant_entitlement(skill.id, "host-1", now=now)
+    # A version older than anything in the changelog must yield a full replace.
+    delta = service.fetch_host_delta("host-1", since_version=-5)
+    assert delta.is_full_replace is True
+
+
+def test_inventory_record_and_readback(service):
+    now = utc_now()
+    payload = json.dumps([{"id": "local-1", "name": "My Note", "origin": "local"}])
+    service.record_host_inventory("host-1", payload, now=now)
+    entry = service.get_host_inventory("host-1")
+    assert entry is not None
+    assert json.loads(entry.payload_json)[0]["name"] == "My Note"

openspec/changes/skill-management-console/tasks.md

@@ -1,9 +1,9 @@
 ## 1. Cloud skill domain, models, and migration
 
-- [ ] 1.1 Add `cloud/skills.py` domain + service layer mirroring `cloud/llm_providers.py`: `CloudSkill` dataclass (id, name, kind, description, tags, content/steps/parameters, version, timestamps), `CloudSkillEntitlement` (skill_id, host_id), validation (`validate_cloud_skill_input`), and a `CloudSkillService` over a repository port. Reuse `skills_learning.models` types where shape aligns; no secrets.
-- [ ] 1.2 Add SQLAlchemy models to `cloud/db_models.py`: `cloud_skills`, `cloud_skill_entitlements`, `cloud_skill_sync_state` (host_id, last_version, updated_at). Index `cloud_skill_entitlements` on (skill_id, host_id) unique.
-- [ ] 1.3 Add Alembic migration `0010_skill_management.py` creating the three tables; downgrade drops them. No existing table altered.
-- [ ] 1.4 Extend `cloud/sql_repository.py` with a `CloudSkillRepository` port + SQL implementation: skill CRUD, entitlement grant/revoke/list-by-host, atomic per-host `entitlement_version` bump on any relevant change, and `fetch_host_delta(host_id, since_version)` returning upserts/removed_ids/latest_version.
+- [x] 1.1 Add `cloud/skills.py` domain + service layer mirroring `cloud/llm_providers.py`: `CloudSkill` dataclass (id, name, kind, description, tags, content/steps/parameters, version, timestamps), `CloudSkillEntitlement` (skill_id, host_id), validation (`validate_skill_input`), and a `CloudSkillService` over a repository port. Reuse `skills_learning.models` types where shape aligns; no secrets.
+- [x] 1.2 Add SQLAlchemy models to `cloud/db_models.py`: `cloud_skills`, `cloud_skill_entitlements`, `cloud_skill_sync_state` (host_id, last_version, updated_at). Index `cloud_skill_entitlements` on (skill_id, host_id) unique.
+- [x] 1.3 Add Alembic migration `0010_skill_management.py` creating the tables (+ per-host changelog + inventory readback); downgrade drops them. No existing table altered.
+- [x] 1.4 Extend `cloud/sql_repository.py` with cloud-skill repository methods: skill CRUD, entitlement grant/revoke/list-by-host, atomic per-host `entitlement_version` bump + changelog on any relevant change, and `fetch_host_delta(host_id, since_version)` returning upserts/removed_ids/latest_version (incremental with full-replace fallback).
 
 ## 2. Cloud admin REST: skill CRUD + entitlement
 

packages/cloud-platform/cloud/db_models.py

@@ -360,3 +360,83 @@ class LlmProviderSettingsRow(Base):
         Integer, nullable=False, default=1, server_default=text("1")
     )
     updated_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class CloudSkillRow(Base):
+    __tablename__ = "cloud_skills"
+    __table_args__ = (
+        UniqueConstraint(
+            "name_normalized",
+            name="uq_cloud_skills_name_normalized",
+        ),
+        Index("ix_cloud_skills_kind", "kind"),
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    name_normalized: Mapped[str] = mapped_column(String, nullable=False)
+    kind: Mapped[str] = mapped_column(String, nullable=False)
+    description: Mapped[str] = mapped_column(Text, nullable=False)
+    tags_json: Mapped[str] = mapped_column(Text, nullable=False)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    steps_json: Mapped[str] = mapped_column(Text, nullable=False)
+    parameters_json: Mapped[str] = mapped_column(Text, nullable=False)
+    revision: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=1, server_default=text("1")
+    )
+    created_at: Mapped[str] = mapped_column(String, nullable=False)
+    updated_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class CloudSkillEntitlementRow(Base):
+    __tablename__ = "cloud_skill_entitlements"
+    __table_args__ = (
+        UniqueConstraint(
+            "skill_id",
+            "host_id",
+            name="uq_cloud_skill_entitlements_skill_host",
+        ),
+        Index("ix_cloud_skill_entitlements_host_id", "host_id"),
+        Index("ix_cloud_skill_entitlements_skill_id", "skill_id"),
+    )
+
+    skill_id: Mapped[str] = mapped_column(
+        String,
+        ForeignKey("cloud_skills.id"),
+        primary_key=True,
+    )
+    host_id: Mapped[str] = mapped_column(String, primary_key=True)
+    granted_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class CloudSkillSyncStateRow(Base):
+    __tablename__ = "cloud_skill_sync_state"
+
+    host_id: Mapped[str] = mapped_column(String, primary_key=True)
+    last_version: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=text("0")
+    )
+    updated_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class CloudSkillHostChangeRow(Base):
+    """Per-host changelog enabling incremental sync (design D3)."""
+
+    __tablename__ = "cloud_skill_host_changes"
+    __table_args__ = (
+        Index("ix_cloud_skill_host_changes_host_version", "host_id", "version"),
+    )
+
+    host_id: Mapped[str] = mapped_column(String, primary_key=True)
+    version: Mapped[int] = mapped_column(Integer, primary_key=True)
+    skill_id: Mapped[str] = mapped_column(String, nullable=False)
+    change_type: Mapped[str] = mapped_column(String, nullable=False)
+    recorded_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class HostSkillInventoryRow(Base):
+    __tablename__ = "host_skill_inventory"
+
+    host_id: Mapped[str] = mapped_column(String, primary_key=True)
+    payload_json: Mapped[str] = mapped_column(Text, nullable=False)
+    reported_at: Mapped[str] = mapped_column(String, nullable=False)

packages/cloud-platform/cloud/migrations/versions/0010_skill_management.py

@@ -0,0 +1,108 @@
+"""Add Cloud-managed Skills, per-host entitlement, and sync versioning."""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0010_skill_management"
+down_revision = "0009_planner_decision_log"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "cloud_skills",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("name_normalized", sa.String(), nullable=False),
+        sa.Column("kind", sa.String(), nullable=False),
+        sa.Column("description", sa.Text(), nullable=False),
+        sa.Column("tags_json", sa.Text(), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("steps_json", sa.Text(), nullable=False),
+        sa.Column("parameters_json", sa.Text(), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("created_at", sa.String(), nullable=False),
+        sa.Column("updated_at", sa.String(), nullable=False),
+        sa.UniqueConstraint(
+            "name_normalized", name="uq_cloud_skills_name_normalized"
+        ),
+    )
+    op.create_index("ix_cloud_skills_kind", "cloud_skills", ["kind"])
+
+    op.create_table(
+        "cloud_skill_entitlements",
+        sa.Column(
+            "skill_id",
+            sa.String(),
+            sa.ForeignKey("cloud_skills.id"),
+            primary_key=True,
+        ),
+        sa.Column("host_id", sa.String(), primary_key=True),
+        sa.Column("granted_at", sa.String(), nullable=False),
+        sa.UniqueConstraint(
+            "skill_id",
+            "host_id",
+            name="uq_cloud_skill_entitlements_skill_host",
+        ),
+    )
+    op.create_index(
+        "ix_cloud_skill_entitlements_host_id", "cloud_skill_entitlements", ["host_id"]
+    )
+    op.create_index(
+        "ix_cloud_skill_entitlements_skill_id",
+        "cloud_skill_entitlements",
+        ["skill_id"],
+    )
+
+    op.create_table(
+        "cloud_skill_sync_state",
+        sa.Column("host_id", sa.String(), primary_key=True),
+        sa.Column("last_version", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("updated_at", sa.String(), nullable=False),
+    )
+
+    op.create_table(
+        "cloud_skill_host_changes",
+        sa.Column("host_id", sa.String(), primary_key=True),
+        sa.Column("version", sa.Integer(), primary_key=True),
+        sa.Column("skill_id", sa.String(), nullable=False),
+        sa.Column("change_type", sa.String(), nullable=False),
+        sa.Column("recorded_at", sa.String(), nullable=False),
+    )
+    op.create_index(
+        "ix_cloud_skill_host_changes_host_version",
+        "cloud_skill_host_changes",
+        ["host_id", "version"],
+    )
+
+    op.create_table(
+        "host_skill_inventory",
+        sa.Column("host_id", sa.String(), primary_key=True),
+        sa.Column("payload_json", sa.Text(), nullable=False),
+        sa.Column("reported_at", sa.String(), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("host_skill_inventory")
+    op.drop_index(
+        "ix_cloud_skill_host_changes_host_version",
+        table_name="cloud_skill_host_changes",
+    )
+    op.drop_table("cloud_skill_host_changes")
+    op.drop_table("cloud_skill_sync_state")
+    op.drop_index(
+        "ix_cloud_skill_entitlements_skill_id",
+        table_name="cloud_skill_entitlements",
+    )
+    op.drop_index(
+        "ix_cloud_skill_entitlements_host_id",
+        table_name="cloud_skill_entitlements",
+    )
+    op.drop_table("cloud_skill_entitlements")
+    op.drop_index("ix_cloud_skills_kind", table_name="cloud_skills")
+    op.drop_table("cloud_skills")

packages/cloud-platform/cloud/schema.py

@@ -9,7 +9,7 @@ from alembic.runtime.migration import MigrationContext
 from cloud.database import create_database_engine, normalize_database_url
 
 
-HEAD_REVISION = "0009_planner_decision_log"
+HEAD_REVISION = "0010_skill_management"
 
 
 class SchemaVersionError(RuntimeError):

packages/cloud-platform/cloud/skills.py

@@ -0,0 +1,225 @@
+"""Domain and service layer for Cloud-managed Skills and per-host entitlement.
+
+Mirrors the structure of :mod:`cloud.llm_providers`: frozen dataclass models,
+explicit validation, and a service that applies invariants above a repository
+port. Per-host entitlement and incremental sync versioning (design D2/D3) live
+here: any change that affects a host's visible skill set advances that host's
+monotonic ``entitlement_version`` and is recorded in a per-host changelog so
+:meth:`fetch_host_delta` can serve a correct incremental delta.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Literal
+from uuid import uuid4
+
+from core.models import utc_now
+
+SkillKind = Literal["knowledge", "flow_template"]
+SUPPORTED_SKILL_KINDS = frozenset({"knowledge", "flow_template"})
+
+
+class CloudSkillValidationError(ValueError):
+    pass
+
+
+class CloudSkillConflictError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class CloudSkill:
+    id: str
+    name: str
+    name_normalized: str
+    kind: SkillKind
+    description: str
+    tags: list[str]
+    content: str  # knowledge-skill body; empty for flow templates
+    steps_json: str  # JSON list of step dicts for flow templates; "[]" otherwise
+    parameters_json: str  # JSON dict for flow templates; "{}" otherwise
+    revision: int  # global content revision, bumps on edit
+    created_at: datetime
+    updated_at: datetime
+
+
+@dataclass(frozen=True)
+class CloudSkillEntitlement:
+    skill_id: str
+    host_id: str
+    granted_at: datetime
+
+
+@dataclass(frozen=True)
+class HostSkillDelta:
+    """Incremental (or full-replace) delta for one host's entitled skills."""
+
+    skills: list[CloudSkill]
+    removed_ids: list[str]
+    latest_version: int
+    is_full_replace: bool
+
+
+@dataclass(frozen=True)
+class HostInventoryEntry:
+    """A host's reported local-skill inventory row (read-only)."""
+
+    host_id: str
+    payload_json: str
+    reported_at: datetime
+
+
+def normalize_name(name: str) -> str:
+    return " ".join(name.split()).casefold()
+
+
+def validate_skill_input(
+    *,
+    name: str,
+    kind: str,
+    description: str,
+    tags: list[str],
+    content: str,
+    steps_json: str,
+    parameters_json: str,
+) -> tuple[str, str, SkillKind]:
+    display = " ".join(name.split())
+    if not display or len(display) > 200:
+        raise CloudSkillValidationError(
+            "Skill name must contain 1 to 200 characters"
+        )
+    if kind not in SUPPORTED_SKILL_KINDS:
+        raise CloudSkillValidationError("Skill kind must be knowledge or flow_template")
+    if len(description) > 2000:
+        raise CloudSkillValidationError("Skill description too long")
+    if len(tags) > 50:
+        raise CloudSkillValidationError("Too many tags")
+    if kind == "knowledge" and not content.strip():
+        raise CloudSkillValidationError("Knowledge skill content must not be empty")
+    return display, normalize_name(display), kind  # type: ignore[return-value]
+
+
+class CloudSkillService:
+    """Applies validation + entitlement-versioning invariants over a repository."""
+
+    def __init__(self, repository: Any) -> None:
+        self._repository = repository
+
+    # -- skill CRUD -------------------------------------------------------
+
+    def list_skills(self) -> list[CloudSkill]:
+        return self._repository.list_cloud_skills()
+
+    def get_skill(self, skill_id: str) -> CloudSkill | None:
+        return self._repository.get_cloud_skill(skill_id)
+
+    def create_skill(
+        self,
+        *,
+        name: str,
+        kind: str,
+        description: str,
+        tags: list[str],
+        content: str,
+        steps_json: str,
+        parameters_json: str,
+        now: datetime,
+    ) -> CloudSkill:
+        display, normalized, validated_kind = validate_skill_input(
+            name=name,
+            kind=kind,
+            description=description,
+            tags=tags,
+            content=content,
+            steps_json=steps_json,
+            parameters_json=parameters_json,
+        )
+        skill = CloudSkill(
+            id=uuid4().hex,
+            name=display,
+            name_normalized=normalized,
+            kind=validated_kind,
+            description=description,
+            tags=list(tags),
+            content=content,
+            steps_json=steps_json,
+            parameters_json=parameters_json,
+            revision=1,
+            created_at=now,
+            updated_at=now,
+        )
+        return self._repository.create_cloud_skill(skill)
+
+    def update_skill(
+        self,
+        skill_id: str,
+        *,
+        name: str,
+        kind: str,
+        description: str,
+        tags: list[str],
+        content: str,
+        steps_json: str,
+        parameters_json: str,
+        now: datetime,
+    ) -> CloudSkill:
+        display, normalized, validated_kind = validate_skill_input(
+            name=name,
+            kind=kind,
+            description=description,
+            tags=tags,
+            content=content,
+            steps_json=steps_json,
+            parameters_json=parameters_json,
+        )
+        return self._repository.update_cloud_skill(
+            skill_id,
+            name=display,
+            name_normalized=normalized,
+            kind=validated_kind,
+            description=description,
+            tags=list(tags),
+            content=content,
+            steps_json=steps_json,
+            parameters_json=parameters_json,
+            updated_at=now,
+        )
+
+    def delete_skill(self, skill_id: str) -> None:
+        # Repository cascades entitlement removal and records a remove-change
+        # for every host that was entitled.
+        self._repository.delete_cloud_skill(skill_id)
+
+    # -- entitlement ------------------------------------------------------
+
+    def list_hosts_for_skill(self, skill_id: str) -> list[str]:
+        return self._repository.list_entitlements_for_skill(skill_id)
+
+    def list_entitled_skills_for_host(self, host_id: str) -> list[CloudSkill]:
+        return self._repository.list_entitled_skills_for_host(host_id)
+
+    def grant_entitlement(self, skill_id: str, host_id: str, now: datetime) -> None:
+        if self._repository.get_cloud_skill(skill_id) is None:
+            raise CloudSkillValidationError("Unknown skill")
+        self._repository.grant_entitlement(skill_id, host_id, now=now)
+
+    def revoke_entitlement(self, skill_id: str, host_id: str, now: datetime) -> None:
+        self._repository.revoke_entitlement(skill_id, host_id, now=now)
+
+    # -- sync -------------------------------------------------------------
+
+    def fetch_host_delta(
+        self, host_id: str, since_version: int | None
+    ) -> HostSkillDelta:
+        return self._repository.fetch_host_delta(host_id, since_version)
+
+    # -- inventory readback -----------------------------------------------
+
+    def record_host_inventory(
+        self, host_id: str, payload_json: str, now: datetime
+    ) -> None:
+        self._repository.record_host_inventory(host_id, payload_json, now=now)
+
+    def get_host_inventory(self, host_id: str) -> HostInventoryEntry | None:
+        return self._repository.get_host_inventory(host_id)

packages/cloud-platform/cloud/sql_repository.py

@@ -30,6 +30,11 @@ from cloud.db_models import (
     UserSubmissionPolicyRow,
     TokenReservationRow,
     TokenUsageEventRow,
+    CloudSkillRow,
+    CloudSkillEntitlementRow,
+    CloudSkillSyncStateRow,
+    CloudSkillHostChangeRow,
+    HostSkillInventoryRow,
 )
 from cloud.observability import current_correlation_id
 from core.models import utc_now
@@ -1772,6 +1777,334 @@ class SQLAlchemyCloudRepository:
     def close(self) -> None:
         self.engine.dispose()
 
+    # ------------------------------------------------------------------
+    # Cloud-managed skills + per-host entitlement + sync versioning
+    # ------------------------------------------------------------------
+
+    def list_cloud_skills(self) -> list[Any]:
+        with self._sessions() as session:
+            rows = session.scalars(
+                select(CloudSkillRow).order_by(CloudSkillRow.name_normalized)
+            ).all()
+        return [_cloud_skill_from_row(row) for row in rows]
+
+    def get_cloud_skill(self, skill_id: str) -> Any | None:
+        with self._sessions() as session:
+            row = session.get(CloudSkillRow, skill_id)
+        return _cloud_skill_from_row(row) if row is not None else None
+
+    def create_cloud_skill(self, skill: Any) -> Any:
+        from cloud.skills import CloudSkillConflictError
+
+        try:
+            with self._sessions.begin() as session:
+                if (
+                    session.scalars(
+                        select(CloudSkillRow).where(
+                            CloudSkillRow.name_normalized == skill.name_normalized
+                        )
+                    ).first()
+                    is not None
+                ):
+                    raise CloudSkillConflictError("Skill name already exists")
+                row = _cloud_skill_to_row(skill)
+                session.add(row)
+                session.flush()
+                return _cloud_skill_from_row(row)
+        except IntegrityError as exc:
+            raise CloudSkillConflictError("Skill name already exists") from exc
+
+    def update_cloud_skill(
+        self,
+        skill_id: str,
+        *,
+        name: str,
+        name_normalized: str,
+        kind: str,
+        description: str,
+        tags: list[str],
+        content: str,
+        steps_json: str,
+        parameters_json: str,
+        updated_at: datetime,
+    ) -> Any:
+        from cloud.skills import CloudSkillConflictError, CloudSkillValidationError
+
+        try:
+            with self._sessions.begin() as session:
+                row = session.get(CloudSkillRow, skill_id)
+                if row is None:
+                    raise CloudSkillValidationError("Unknown skill")
+                clash = session.scalars(
+                    select(CloudSkillRow).where(
+                        CloudSkillRow.name_normalized == name_normalized,
+                        CloudSkillRow.id != skill_id,
+                    )
+                ).first()
+                if clash is not None:
+                    raise CloudSkillConflictError("Skill name already exists")
+                row.name = name
+                row.name_normalized = name_normalized
+                row.kind = kind
+                row.description = description
+                row.tags_json = json.dumps(list(tags))
+                row.content = content
+                row.steps_json = steps_json
+                row.parameters_json = parameters_json
+                row.revision = row.revision + 1
+                row.updated_at = _iso(updated_at)
+                session.flush()
+                updated = _cloud_skill_from_row(row)
+                # Notify every entitled host that the skill changed.
+                host_ids = [
+                    row.host_id
+                    for row in session.scalars(
+                        select(CloudSkillEntitlementRow).where(
+                            CloudSkillEntitlementRow.skill_id == skill_id
+                        )
+                    ).all()
+                ]
+                for host_id in host_ids:
+                    self._bump_host(session, host_id, skill_id, "upsert", updated_at)
+                return updated
+        except IntegrityError as exc:
+            raise CloudSkillConflictError("Skill name already exists") from exc
+
+    def delete_cloud_skill(self, skill_id: str) -> None:
+        with self._sessions.begin() as session:
+            host_ids = [
+                row.host_id
+                for row in session.scalars(
+                    select(CloudSkillEntitlementRow).where(
+                        CloudSkillEntitlementRow.skill_id == skill_id
+                    )
+                ).all()
+            ]
+            now = utc_now()
+            for host_id in host_ids:
+                self._bump_host(session, host_id, skill_id, "remove", now)
+            session.execute(
+                delete(CloudSkillEntitlementRow).where(
+                    CloudSkillEntitlementRow.skill_id == skill_id
+                )
+            )
+            session.execute(
+                delete(CloudSkillRow).where(CloudSkillRow.id == skill_id)
+            )
+
+    def list_entitlements_for_skill(self, skill_id: str) -> list[str]:
+        with self._sessions() as session:
+            rows = session.scalars(
+                select(CloudSkillEntitlementRow).where(
+                    CloudSkillEntitlementRow.skill_id == skill_id
+                )
+            ).all()
+        return [row.host_id for row in rows]
+
+    def list_entitled_skills_for_host(self, host_id: str) -> list[Any]:
+        with self._sessions() as session:
+            entitlements = session.scalars(
+                select(CloudSkillEntitlementRow).where(
+                    CloudSkillEntitlementRow.host_id == host_id
+                )
+            ).all()
+            skills: list[Any] = []
+            for ent in entitlements:
+                row = session.get(CloudSkillRow, ent.skill_id)
+                if row is not None:
+                    skills.append(_cloud_skill_from_row(row))
+        skills.sort(key=lambda s: s.name_normalized)
+        return skills
+
+    def grant_entitlement(
+        self, skill_id: str, host_id: str, *, now: datetime
+    ) -> None:
+        with self._sessions.begin() as session:
+            existing = session.get(
+                CloudSkillEntitlementRow, (skill_id, host_id)
+            )
+            if existing is not None:
+                return  # idempotent
+            session.add(
+                CloudSkillEntitlementRow(
+                    skill_id=skill_id,
+                    host_id=host_id,
+                    granted_at=_iso(now),
+                )
+            )
+            self._bump_host(session, host_id, skill_id, "upsert", now)
+
+    def revoke_entitlement(
+        self, skill_id: str, host_id: str, *, now: datetime
+    ) -> None:
+        with self._sessions.begin() as session:
+            existing = session.get(
+                CloudSkillEntitlementRow, (skill_id, host_id)
+            )
+            if existing is None:
+                return
+            session.delete(existing)
+            self._bump_host(session, host_id, skill_id, "remove", now)
+
+    def fetch_host_delta(
+        self, host_id: str, since_version: int | None
+    ) -> Any:
+        from cloud.skills import HostSkillDelta
+
+        with self._sessions.begin() as session:
+            state = self._ensure_sync_state(session, host_id)
+            latest = state.last_version
+            if since_version is None:
+                skills = self._host_skills_in_session(session, host_id)
+                return HostSkillDelta(
+                    skills=skills,
+                    removed_ids=[],
+                    latest_version=latest,
+                    is_full_replace=True,
+                )
+            oldest = self._oldest_changelog_version(session, host_id)
+            if oldest is None or since_version < oldest - 1:
+                skills = self._host_skills_in_session(session, host_id)
+                return HostSkillDelta(
+                    skills=skills,
+                    removed_ids=[],
+                    latest_version=latest,
+                    is_full_replace=True,
+                )
+            changes = (
+                session.scalars(
+                    select(CloudSkillHostChangeRow)
+                    .where(
+                        CloudSkillHostChangeRow.host_id == host_id,
+                        CloudSkillHostChangeRow.version > since_version,
+                    )
+                    .order_by(CloudSkillHostChangeRow.version)
+                )
+                .unique()
+                .all()
+            )
+            upsert_ids: list[str] = []
+            removed_ids: list[str] = []
+            seen: set[str] = set()
+            for change in changes:
+                if change.skill_id in seen:
+                    continue
+                seen.add(change.skill_id)
+                if change.change_type == "remove":
+                    removed_ids.append(change.skill_id)
+                else:
+                    upsert_ids.append(change.skill_id)
+            skills: list[Any] = []
+            entitled_ids = {
+                ent.skill_id
+                for ent in session.scalars(
+                    select(CloudSkillEntitlementRow).where(
+                        CloudSkillEntitlementRow.host_id == host_id
+                    )
+                ).all()
+            }
+            for sid in upsert_ids:
+                if sid in entitled_ids:
+                    row = session.get(CloudSkillRow, sid)
+                    if row is not None:
+                        skills.append(_cloud_skill_from_row(row))
+            skills.sort(key=lambda s: s.name_normalized)
+            return HostSkillDelta(
+                skills=skills,
+                removed_ids=removed_ids,
+                latest_version=latest,
+                is_full_replace=False,
+            )
+
+    def record_host_inventory(
+        self, host_id: str, payload_json: str, *, now: datetime
+    ) -> None:
+        with self._sessions.begin() as session:
+            row = session.get(HostSkillInventoryRow, host_id)
+            if row is None:
+                session.add(
+                    HostSkillInventoryRow(
+                        host_id=host_id,
+                        payload_json=payload_json,
+                        reported_at=_iso(now),
+                    )
+                )
+            else:
+                row.payload_json = payload_json
+                row.reported_at = _iso(now)
+
+    def get_host_inventory(self, host_id: str) -> Any | None:
+        from cloud.skills import HostInventoryEntry
+
+        with self._sessions() as session:
+            row = session.get(HostSkillInventoryRow, host_id)
+        if row is None:
+            return None
+        return HostInventoryEntry(
+            host_id=row.host_id,
+            payload_json=row.payload_json,
+            reported_at=_parse_dt(row.reported_at),
+        )
+
+    # -- internals --------------------------------------------------------
+
+    def _ensure_sync_state(self, session: Any, host_id: str) -> Any:
+        row = session.get(CloudSkillSyncStateRow, host_id)
+        if row is None:
+            row = CloudSkillSyncStateRow(
+                host_id=host_id,
+                last_version=0,
+                updated_at=_iso(utc_now()),
+            )
+            session.add(row)
+            session.flush()
+        return row
+
+    def _bump_host(
+        self,
+        session: Any,
+        host_id: str,
+        skill_id: str,
+        change_type: str,
+        now: datetime,
+    ) -> None:
+        state = self._ensure_sync_state(session, host_id)
+        state.last_version = state.last_version + 1
+        state.updated_at = _iso(now)
+        session.add(
+            CloudSkillHostChangeRow(
+                host_id=host_id,
+                version=state.last_version,
+                skill_id=skill_id,
+                change_type=change_type,
+                recorded_at=_iso(now),
+            )
+        )
+
+    def _oldest_changelog_version(self, session: Any, host_id: str) -> int | None:
+        row = session.scalars(
+            select(func.min(CloudSkillHostChangeRow.version)).where(
+                CloudSkillHostChangeRow.host_id == host_id
+            )
+        ).first()
+        return int(row) if row is not None else None
+
+    def _host_skills_in_session(self, session: Any, host_id: str) -> list[Any]:
+        entitlements = session.scalars(
+            select(CloudSkillEntitlementRow).where(
+                CloudSkillEntitlementRow.host_id == host_id
+            )
+        ).all()
+        skills = [
+            _cloud_skill_from_row(row)
+            for row in (
+                session.get(CloudSkillRow, ent.skill_id) for ent in entitlements
+            )
+            if row is not None
+        ]
+        skills.sort(key=lambda s: s.name_normalized)
+        return skills
+
 
 def _iso(value: datetime) -> str:
     return value.isoformat()
@@ -2154,3 +2487,40 @@ def _llm_provider_settings_from_row(row: LlmProviderSettingsRow | None) -> Any:
         revision=row.revision,
         updated_at=_parse_dt(row.updated_at),
     )
+
+
+def _cloud_skill_from_row(row: CloudSkillRow) -> Any:
+    from cloud.skills import CloudSkill
+
+    now = utc_now()
+    return CloudSkill(
+        id=row.id,
+        name=row.name,
+        name_normalized=row.name_normalized,
+        kind=row.kind,  # type: ignore[arg-type]
+        description=row.description,
+        tags=json.loads(row.tags_json or "[]"),
+        content=row.content,
+        steps_json=row.steps_json,
+        parameters_json=row.parameters_json,
+        revision=row.revision,
+        created_at=_parse_dt(row.created_at) or now,
+        updated_at=_parse_dt(row.updated_at) or now,
+    )
+
+
+def _cloud_skill_to_row(skill: Any) -> CloudSkillRow:
+    return CloudSkillRow(
+        id=skill.id,
+        name=skill.name,
+        name_normalized=skill.name_normalized,
+        kind=skill.kind,
+        description=skill.description,
+        tags_json=json.dumps(list(skill.tags)),
+        content=skill.content,
+        steps_json=skill.steps_json,
+        parameters_json=skill.parameters_json,
+        revision=skill.revision,
+        created_at=_iso(skill.created_at),
+        updated_at=_iso(skill.updated_at),
+    )