agentic-mobile-control/api/skill_catalog_view.py

"""Unified read-merge surface over the synced catalog and the local skill store.

API layer per CONSTITUTION.md: this module reads from both
:mod:`storage.skill_catalog` (cloud-synced, read-only-except-sync) and
:mod:`storage.local_skills` (agent-authored + overrides) and presents a single
origin-discriminated catalog. It is the only module that reads both stores.

Override semantics (design D8/D9/D11): a local override shadows the cloud
skill at read time, wins against sync updates, and is reported with
``origin = "cloud"`` and ``locally_overridden = true`` while it shadows.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Literal

from skills_learning.models import (
    FlowTemplateSkill,
    KnowledgeSkill,
    SkillMetadata,
)
from storage.local_skills import LocalSkillStore
from storage.skill_catalog import SkillCatalogStore

Origin = Literal["cloud", "local"]


@dataclass(frozen=True)
class SkillSummary:
    """Metadata + origin for list/search results."""

    metadata: SkillMetadata
    origin: Origin
    locally_overridden: bool


@dataclass(frozen=True)
class SkillView:
    """Full skill + origin for get_skill results."""

    skill: KnowledgeSkill | FlowTemplateSkill
    origin: Origin
    locally_overridden: bool


class SkillCatalogView:
    """Merged read surface over synced + local skills, with override precedence."""

    def __init__(self, synced: SkillCatalogStore, local: LocalSkillStore) -> None:
        self._synced = synced
        self._local = local

    # ------------------------------------------------------------------
    # Merged reads
    # ------------------------------------------------------------------

    def list_skills(
        self,
        active_subscriptions: set[str],
    ) -> list[SkillSummary]:
        override_ids = self._local.list_override_cloud_ids()
        summaries: list[SkillSummary] = []

        # Local authored skills.
        for meta in self._local.list_local():
            summaries.append(SkillSummary(meta, "local", False))

        # Cloud-synced skills, applying overrides where present.
        for meta in self._synced.list_skills(active_subscriptions):
            if meta.id in override_ids:
                override = self._local.get_override(meta.id)
                if override is not None:
                    summaries.append(SkillSummary(override.metadata, "cloud", True))
                    continue
            summaries.append(SkillSummary(meta, "cloud", False))

        summaries.sort(key=lambda s: s.metadata.name)
        return summaries

    def search_skills(
        self,
        query: str,
        active_subscriptions: set[str],
    ) -> list[SkillSummary]:
        override_ids = self._local.list_override_cloud_ids()
        summaries: list[SkillSummary] = []

        for meta in self._local.search_local(query):
            summaries.append(SkillSummary(meta, "local", False))

        for meta in self._synced.search_skills(query, active_subscriptions):
            if meta.id in override_ids:
                override = self._local.get_override(meta.id)
                if override is not None:
                    summaries.append(SkillSummary(override.metadata, "cloud", True))
                    continue
            summaries.append(SkillSummary(meta, "cloud", False))

        return summaries

    def get_skill(
        self,
        skill_id: str,
        active_subscriptions: set[str],
        *,
        registered_tools: set[str] | None = None,
    ) -> SkillView | None:
        # 1. Local authored skill.
        local_skill = self._local.get_local(skill_id)
        if local_skill is not None:
            return SkillView(local_skill, "local", False)

        # 2. Override shadowing a cloud skill id.
        override = self._local.get_override(skill_id)
        if override is not None:
            if (
                isinstance(override, FlowTemplateSkill)
                and registered_tools is not None
                and not _tools_valid(override, registered_tools)
            ):
                return None
            return SkillView(override, "cloud", True)

        # 3. Cloud-synced skill (None for unknown AND not-visible — no leak).
        cloud_skill = self._synced.get_skill(
            skill_id,
            active_subscriptions,
            registered_tools=registered_tools,
        )
        if cloud_skill is not None:
            return SkillView(cloud_skill, "cloud", False)
        return None

    # ------------------------------------------------------------------
    # Origin classification for authoring dispatch (D10)
    # ------------------------------------------------------------------

    def is_local_skill(self, skill_id: str) -> bool:
        return self._local.get_local(skill_id) is not None

    def has_override(self, cloud_skill_id: str) -> bool:
        return self._local.has_override(cloud_skill_id)


def _tools_valid(skill: FlowTemplateSkill, registered_tools: set[str]) -> bool:
    return all(step.tool_name in registered_tools for step in skill.steps)


def make_view_from_stores(
    synced: SkillCatalogStore,
    local: LocalSkillStore,
) -> SkillCatalogView:
    return SkillCatalogView(synced, local)


def default_local_store() -> LocalSkillStore:
    """Lazy default local store (created on first use)."""
    return LocalSkillStore()