agentic-mobile-control/packages/cloud-platform/cloud/skills.py

"""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
from datetime import datetime
from typing import Any, Literal
from uuid import uuid4


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)