From 52e442790a8745604f68557c701656da92d2e8db Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 15 Jul 2026 07:45:22 +0800 Subject: [PATCH] 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 --- .../tests/test_cloud_skill_management.py | 143 +++++++ .../changes/skill-management-console/tasks.md | 8 +- packages/cloud-platform/cloud/db_models.py | 80 ++++ .../versions/0010_skill_management.py | 108 +++++ packages/cloud-platform/cloud/schema.py | 2 +- packages/cloud-platform/cloud/skills.py | 225 +++++++++++ .../cloud-platform/cloud/sql_repository.py | 370 ++++++++++++++++++ 7 files changed, 931 insertions(+), 5 deletions(-) create mode 100644 apps/cloud-api/tests/test_cloud_skill_management.py create mode 100644 packages/cloud-platform/cloud/migrations/versions/0010_skill_management.py create mode 100644 packages/cloud-platform/cloud/skills.py diff --git a/apps/cloud-api/tests/test_cloud_skill_management.py b/apps/cloud-api/tests/test_cloud_skill_management.py new file mode 100644 index 0000000..25fd30d --- /dev/null +++ b/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" diff --git a/openspec/changes/skill-management-console/tasks.md b/openspec/changes/skill-management-console/tasks.md index 51707a9..bef4578 100644 --- a/openspec/changes/skill-management-console/tasks.md +++ b/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 diff --git a/packages/cloud-platform/cloud/db_models.py b/packages/cloud-platform/cloud/db_models.py index c5acbef..3490271 100644 --- a/packages/cloud-platform/cloud/db_models.py +++ b/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) diff --git a/packages/cloud-platform/cloud/migrations/versions/0010_skill_management.py b/packages/cloud-platform/cloud/migrations/versions/0010_skill_management.py new file mode 100644 index 0000000..2796043 --- /dev/null +++ b/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") diff --git a/packages/cloud-platform/cloud/schema.py b/packages/cloud-platform/cloud/schema.py index e9fb4de..c38d932 100644 --- a/packages/cloud-platform/cloud/schema.py +++ b/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): diff --git a/packages/cloud-platform/cloud/skills.py b/packages/cloud-platform/cloud/skills.py new file mode 100644 index 0000000..e557390 --- /dev/null +++ b/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) diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index c7150f7..998bddd 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/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), + )