writer-work-flow/packages/skills/ww_skills/skill_registry.py

"""Skill registry loader（ARCH §5.6 加载段 / PRODUCT_SPEC §5.5）。

从 `SkillRepo`（`skills` 表的读侧抽象）加载声明式 skill 行 → 转 frozen `AgentSpec`
（内置 builtin / 用户 custom / 社区 community 同构）。加载后按 `tier` 经 LLM 网关执行
（复用 §5.1 的 agent 机制，本模块**不**重实现网关）。

加载时对每条声明跑 `validate_declaration`：越权（reads/writes 指向未知表）的 skill
直接拒绝（不入册），守 §5.6 表权限契约（不可信用户输入的第一道闸）。

不变量 #2：行里只带 `tier`（writer/analyst/light），registry 不解析具体 model
（model 解析在网关）。registry 加载后不可变（specs 字典只读）。
"""

from __future__ import annotations

from typing import Protocol, cast, get_args

from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ww_agents import REVIEW_RESERVED_NAMES, SPECS, AgentSpec
from ww_db.models import Skill
from ww_llm_gateway.types import Tier
from ww_shared import AppError, ErrorCode

from ww_skills.skill_permissions import validate_declaration

_VALID_TIERS: frozenset[str] = frozenset(get_args(Tier))

# 内置保留命名空间：四审受信名（REVIEW_RESERVED_NAMES）∪ 全部内置 SPECS name。
# 用户 skill 不可冒用这些 name（守卫前移在入库期，呼应不变量 #3：四审 prompt 不被偷换）。
_RESERVED_NAMES: frozenset[str] = REVIEW_RESERVED_NAMES | frozenset(SPECS)


def _reject_reserved_name(spec: AgentSpec) -> None:
    """用户 skill 冒用内置/保留名 → AppError(VALIDATION)（不可信声明的第一道闸）。"""
    if spec.name in _RESERVED_NAMES:
        raise AppError(
            ErrorCode.VALIDATION,
            f"Skill「{spec.name}」与内置保留名冲突，不可覆盖",
            {"skill": spec.name},
        )


class SkillRecord(BaseModel):
    """`skills` 表一行的声明式快照（frozen；snake_case）。

    `input_schema`/`output_schema` 在 DB 里是 JSON Schema dict（用户自定义无 Python 类型），
    registry 暂以纯声明（prompt + 表权限 + tier）执行；结构化 schema 的运行期绑定属后续，
    故这里不携带 type[BaseModel]（内置 8 agent 仍用各自硬编码的 spec，见 ww_agents）。
    """

    model_config = {"frozen": True}

    name: str
    scope: str  # builtin / custom / community
    tier: Tier
    system_prompt: str
    # 不可变序列：frozen 只防整字段重绑，不防 list 原地变异（record.reads.append 可旁路）；
    # tuple 无 append/clear，对齐 AgentSpec 同字段语义。Pydantic v2 coerce list→tuple。
    reads: tuple[str, ...] = ()
    writes: tuple[str, ...] = ()
    genre: str | None = None


def _to_spec(record: SkillRecord) -> AgentSpec:
    return AgentSpec(
        name=record.name,
        tier=record.tier,
        system_prompt=record.system_prompt,
        input_schema=None,
        output_schema=None,
        reads=record.reads,
        writes=record.writes,
        genre=record.genre,
        scope=record.scope,
    )


class SkillRepo(Protocol):
    """`skills` 表读侧接口（registry 加载源；测试注内存 fake）。"""

    async def list_all(self) -> list[SkillRecord]: ...


class SkillRegistryLike(Protocol):
    """`SpecResolver` 消费的 registry 只读契约（结构子集；测试注 duck-type fake）。

    SpecResolver 只用 `.get` / `.names` / `.list_scope` 三个方法——以 Protocol 精确捕获该
    契约，避免依赖具体 `SkillRegistry` 类逼测试 fake 加 `# type: ignore`。
    """

    def get(self, name: str) -> AgentSpec: ...

    def names(self) -> list[str]: ...

    def list_scope(self, scope: str) -> list[AgentSpec]: ...


class SkillRegistry:
    """加载后的只读 skill 注册表（name → AgentSpec）。"""

    def __init__(self, specs: dict[str, AgentSpec]) -> None:
        self._specs = dict(specs)

    @classmethod
    async def load(cls, repo: SkillRepo) -> SkillRegistry:
        """从 repo 拉全部声明 → 校验表权限 → 建 frozen AgentSpec 字典。

        越权声明（reads/writes 指向未知表）→ AppError(VALIDATION)，加载中断（不入册）。
        """
        specs: dict[str, AgentSpec] = {}
        for record in await repo.list_all():
            spec = _to_spec(record)
            _reject_reserved_name(spec)  # 冒用内置/保留名 → 抛 VALIDATION
            validate_declaration(spec)  # 越权 → 抛 VALIDATION
            specs[spec.name] = spec
        return cls(specs)

    def get(self, name: str) -> AgentSpec:
        """按名取 spec；不存在 → AppError(NOT_FOUND)。"""
        spec = self._specs.get(name)
        if spec is None:
            raise AppError(ErrorCode.NOT_FOUND, f"skill not found: {name}")
        return spec

    def names(self) -> list[str]:
        return sorted(self._specs)

    def list_scope(self, scope: str) -> list[AgentSpec]:
        """按 scope（builtin/custom/community）过滤，按 name 排序。"""
        return [s for s in sorted(self._specs.values(), key=lambda s: s.name) if s.scope == scope]


def _row_to_record(row: Skill) -> SkillRecord:
    # DB `tier` 是裸 str，非法档位 → VALIDATION（不可信用户输入第一道闸）。
    if row.tier not in _VALID_TIERS:
        raise AppError(
            ErrorCode.VALIDATION,
            f"Skill「{row.name}」声明了非法档位：{row.tier}",
            {"skill": row.name, "tier": row.tier},
        )
    return SkillRecord(
        name=row.name,
        scope=row.scope,
        tier=cast(Tier, row.tier),
        system_prompt=row.system_prompt,
        reads=tuple(str(t) for t in (row.reads or [])),
        writes=tuple(str(t) for t in (row.writes or [])),
        genre=row.genre,
    )


class SqlSkillRepo:
    """`skills` 表读侧 SQLAlchemy 实现（registry 加载源；只读）。"""

    def __init__(self, session: AsyncSession) -> None:
        self._s = session

    async def list_all(self) -> list[SkillRecord]:
        rows = (await self._s.execute(select(Skill))).scalars().all()
        return [_row_to_record(r) for r in rows]