oclaw/runtime/skill_manifest_core.py

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class SkillInstallSpec:
    id: str
    kind: str
    payload: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class SkillRuntimeSpec:
    type: str
    entry: str
    schema: dict[str, Any] = field(default_factory=dict)
    permissions: dict[str, Any] = field(default_factory=dict)
    timeout_s: float | None = None
    max_output_bytes: int | None = None

    def as_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"type": self.type, "entry": self.entry}
        if self.schema:
            out["schema"] = dict(self.schema)
        if self.permissions:
            out["permissions"] = dict(self.permissions)
        if isinstance(self.timeout_s, (int, float)):
            out["timeout_s"] = float(self.timeout_s)
        if isinstance(self.max_output_bytes, int):
            out["max_output_bytes"] = int(self.max_output_bytes)
        return out


@dataclass(frozen=True)
class ParsedSkillFrontmatter:
    name: str
    description: str
    user_invocable: bool
    disable_model_invocation: bool
    metadata_oclaw: dict[str, Any]
    install: tuple[SkillInstallSpec, ...]
    runtime: SkillRuntimeSpec | None


def normalize_frontmatter(fm: dict[str, Any]) -> dict[str, Any]:
    out = dict(fm)
    md = out.get("metadata")
    if isinstance(md, str) and md.strip():
        try:
            out["metadata"] = json.loads(md)
        except Exception:
            out["metadata"] = {}
    elif not isinstance(md, dict):
        out["metadata"] = {}
    for key, default in (("user-invocable", True), ("disable-model-invocation", False)):
        val = out.get(key, default)
        if isinstance(val, bool):
            out[key] = val
        else:
            out[key] = str(val or str(default)).strip().lower() in {"1", "true", "yes", "on"}
    return out


def parse_install_specs(metadata_oclaw: dict[str, Any]) -> tuple[SkillInstallSpec, ...]:
    raw = metadata_oclaw.get("install")
    if not isinstance(raw, list):
        return ()
    out: list[SkillInstallSpec] = []
    for idx, it in enumerate(raw):
        if not isinstance(it, dict):
            continue
        sid = str(it.get("id") or f"install_{idx + 1}").strip() or f"install_{idx + 1}"
        kind = str(it.get("kind") or "").strip().lower()
        if not kind:
            continue
        out.append(SkillInstallSpec(id=sid, kind=kind, payload=dict(it)))
    return tuple(out)


def parse_runtime_spec(metadata_oclaw: dict[str, Any]) -> SkillRuntimeSpec | None:
    raw = metadata_oclaw.get("runtime")
    if not isinstance(raw, dict):
        return None
    tp = str(raw.get("type") or "").strip().lower()
    if tp not in {"shell", "python", "node", "hook"}:
        return None
    entry = str(raw.get("entry") or "").strip().replace("\\", "/")
    if not entry or entry.startswith("/") or ".." in entry.split("/"):
        return None
    schema = raw.get("schema") if isinstance(raw.get("schema"), dict) else {}
    perms = raw.get("permissions") if isinstance(raw.get("permissions"), dict) else {}
    timeout_s_raw = raw.get("timeout_s")
    timeout_s = float(timeout_s_raw) if isinstance(timeout_s_raw, (int, float)) else None
    max_output_raw = raw.get("max_output_bytes")
    max_output = int(max_output_raw) if isinstance(max_output_raw, int) else None
    return SkillRuntimeSpec(
        type=tp,
        entry=entry,
        schema=dict(schema),
        permissions=dict(perms),
        timeout_s=timeout_s,
        max_output_bytes=max_output,
    )


def parse_skill_frontmatter(*, fm: dict[str, Any], default_name: str) -> ParsedSkillFrontmatter:
    nfm = normalize_frontmatter(fm)
    name = str(nfm.get("name") or default_name).strip() or default_name
    desc = str(nfm.get("description") or "").strip() or f"{name} skill"
    md = nfm.get("metadata")
    md = dict(md) if isinstance(md, dict) else {}
    oc = md.get("oclaw")
    oc = dict(oc) if isinstance(oc, dict) else {}
    return ParsedSkillFrontmatter(
        name=name,
        description=desc,
        user_invocable=bool(nfm.get("user-invocable", True)),
        disable_model_invocation=bool(nfm.get("disable-model-invocation", False)),
        metadata_oclaw=oc,
        install=parse_install_specs(oc),
        runtime=parse_runtime_spec(oc),
    )


__all__ = [
    "SkillInstallSpec",
    "SkillRuntimeSpec",
    "ParsedSkillFrontmatter",
    "normalize_frontmatter",
    "parse_install_specs",
    "parse_runtime_spec",
    "parse_skill_frontmatter",
]