agent-first-data 0.17.1

"""Reusable Agent Skill installer for spore CLIs.

A spore that embeds its ``SKILL.md`` describes itself with a :class:`SkillSpec` and
calls :func:`run_skill_admin` to install, uninstall, or report status of that skill
across supported coding agents (Codex, Claude Code, opencode, Hermes).

The function performs the filesystem work and returns the protocol ``dict`` (rendered
by the caller with ``cli_output``) or raises :class:`SkillError`. It never writes to
stdout/stderr itself.
"""

from __future__ import annotations

import enum
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional, Union

SKILL_FILE_NAME = "SKILL.md"
_FNV1A64_OFFSET = 0xCBF29CE484222325
_FNV1A64_PRIME = 0x100000001B3
_FNV1A64_MASK = 0xFFFFFFFFFFFFFFFF


@dataclass(frozen=True)
class SkillSpec:
    """Identity of the skill being managed and the tool that manages it.

    ``name`` is both the skill directory name and the ``name:`` front-matter field.
    ``source`` is the bundled ``SKILL.md``. ``title`` is a human label for error
    messages. ``marker_slug`` seeds the managed-skill marker and the
    ``Generated by <slug> skill install`` comment, and is referenced in hints.
    """

    name: str
    source: str
    title: str
    marker_slug: str


class SkillAction(enum.Enum):
    """The skill action to perform."""

    STATUS = "status"
    INSTALL = "install"
    UNINSTALL = "uninstall"


class SkillScope(enum.Enum):
    """Where to install the skill."""

    PERSONAL = "personal"
    WORKSPACE = "workspace"


class SkillAgentSelection(enum.Enum):
    """Which agent target(s) to manage."""

    ALL = "all"
    CODEX = "codex"
    CLAUDE_CODE = "claude-code"
    OPENCODE = "opencode"
    HERMES = "hermes"


class SkillAgent(enum.Enum):
    """A concrete agent a skill is installed for (no ``all``)."""

    CODEX = "codex"
    CLAUDE_CODE = "claude-code"
    OPENCODE = "opencode"
    HERMES = "hermes"


@dataclass
class SkillOptions:
    """Options shared by every skill action."""

    agent: SkillAgentSelection
    scope: SkillScope
    skills_dir: Optional[str] = None
    force: bool = False


class SkillError(Exception):
    """A skill admin failure with an operator-facing message and optional hint."""

    def __init__(self, message: str, hint: Optional[str] = None) -> None:
        super().__init__(message)
        self.message = message
        self.hint = hint


@dataclass
class SkillTargetStatus:
    """Per-target outcome of ``status`` / ``install``."""

    agent: SkillAgent
    scope: SkillScope
    skills_dir: str
    skill_path: str
    installed: bool
    managed: bool
    valid: bool
    current: bool
    validation_error: Optional[str]

    def to_dict(self) -> dict:
        return {
            "agent": self.agent.value,
            "scope": self.scope.value,
            "skills_dir": self.skills_dir,
            "skill_path": self.skill_path,
            "installed": self.installed,
            "managed": self.managed,
            "valid": self.valid,
            "current": self.current,
            "validation_error": self.validation_error,
        }


@dataclass
class SkillUninstallStatus:
    """Per-target outcome of ``uninstall``."""

    agent: SkillAgent
    scope: SkillScope
    skills_dir: str
    skill_path: str
    removed: bool

    def to_dict(self) -> dict:
        return {
            "agent": self.agent.value,
            "scope": self.scope.value,
            "skills_dir": self.skills_dir,
            "skill_path": self.skill_path,
            "removed": self.removed,
        }


@dataclass
class SkillStatusReport:
    """The result of a ``status`` action."""

    skill: str
    installed_all: bool
    valid_all: bool
    current_all: bool
    targets: List[SkillTargetStatus]
    code: str = "skill_status"

    def to_dict(self) -> dict:
        return {
            "code": self.code,
            "skill": self.skill,
            "installed_all": self.installed_all,
            "valid_all": self.valid_all,
            "current_all": self.current_all,
            "targets": [t.to_dict() for t in self.targets],
        }


@dataclass
class SkillInstallReport:
    """The result of an ``install`` action."""

    skill: str
    targets: List[SkillTargetStatus]
    installed: bool = True
    hint: str = "restart the agent so it reloads installed skills"
    code: str = "skill_install"

    def to_dict(self) -> dict:
        return {
            "code": self.code,
            "skill": self.skill,
            "installed": self.installed,
            "targets": [t.to_dict() for t in self.targets],
            "hint": self.hint,
        }


@dataclass
class SkillUninstallReport:
    """The result of an ``uninstall`` action."""

    skill: str
    removed_any: bool
    targets: List[SkillUninstallStatus]
    code: str = "skill_uninstall"

    def to_dict(self) -> dict:
        return {
            "code": self.code,
            "skill": self.skill,
            "removed_any": self.removed_any,
            "targets": [t.to_dict() for t in self.targets],
        }


# A skill action result. Read fields directly, or call ``to_dict()`` to serialize
# with ``cli_output``. The ``code`` field is the protocol discriminator.
SkillReport = Union[SkillStatusReport, SkillInstallReport, SkillUninstallReport]


@dataclass
class _SkillTarget:
    agent: SkillAgent
    scope: SkillScope
    skills_dir: Path
    skill_dir: Path
    skill_path: Path


def run_skill_admin(spec: SkillSpec, action: SkillAction, options: SkillOptions) -> SkillReport:
    """Install, uninstall, or report status of ``spec``'s skill across the selected
    agent target(s). Returns a typed :data:`SkillReport` (read fields directly, or
    call ``to_dict()`` to serialize with ``cli_output``) or raises :class:`SkillError`.
    Does not touch stdout/stderr.
    """
    _validate_spec(spec)
    if action is SkillAction.STATUS:
        return _status(spec, options)
    if action is SkillAction.INSTALL:
        return _install(spec, options)
    if action is SkillAction.UNINSTALL:
        return _uninstall(spec, options)
    raise SkillError(f"unknown skill action {action!r}")


def _status(spec: SkillSpec, options: SkillOptions) -> SkillStatusReport:
    targets = _resolve_targets(spec, options)
    statuses = [_target_status(spec, target) for target in targets]
    return SkillStatusReport(
        skill=spec.name,
        installed_all=all(s.installed for s in statuses),
        valid_all=all(s.valid for s in statuses),
        current_all=all(s.current for s in statuses),
        targets=statuses,
    )


def _install(spec: SkillSpec, options: SkillOptions) -> SkillInstallReport:
    _validate_skill_text(spec, spec.source)
    targets = _resolve_targets(spec, options)
    content = _managed_skill_contents(spec)
    installed = []
    for target in targets:
        target.skill_dir.mkdir(parents=True, exist_ok=True)
        if target.skill_path.is_symlink():
            if not options.force:
                raise SkillError(
                    f"refusing to overwrite symlinked skill at {target.skill_path}",
                    "pass --force to replace the symlink itself, or choose another --skills-dir",
                )
        elif target.skill_path.exists() and not _is_managed_or_bundled_skill(spec, target.skill_path) and not options.force:
            raise SkillError(
                f"refusing to overwrite unmanaged skill at {target.skill_path}",
                "pass --force to replace it, or choose another --skills-dir",
            )
        _write_skill_atomic(target, content)
        _validate_installed_skill(spec, target.skill_path)
        installed.append(_target_status(spec, target))
    return SkillInstallReport(skill=spec.name, targets=installed)


def _uninstall(spec: SkillSpec, options: SkillOptions) -> SkillUninstallReport:
    targets = _resolve_targets(spec, options)
    removed = []
    for target in targets:
        if not target.skill_path.exists() and not target.skill_path.is_symlink():
            removed.append(_target_uninstall_status(target, False))
            continue
        if target.skill_path.is_symlink():
            if not options.force:
                raise SkillError(
                    f"refusing to remove symlinked skill at {target.skill_path}",
                    "pass --force to remove the symlink itself",
                )
        elif not _is_managed_or_bundled_skill(spec, target.skill_path) and not options.force:
            raise SkillError(
                f"refusing to remove unmanaged skill at {target.skill_path}",
                f"only skills generated by {spec.marker_slug} skill install can be removed without --force",
            )
        target.skill_path.unlink()
        # Prune the now-empty skill dir; ignore if not empty.
        try:
            target.skill_dir.rmdir()
        except OSError:
            pass
        removed.append(_target_uninstall_status(target, True))
    return SkillUninstallReport(
        skill=spec.name,
        removed_any=any(s.removed for s in removed),
        targets=removed,
    )


def _resolve_targets(spec: SkillSpec, options: SkillOptions) -> list[_SkillTarget]:
    if options.skills_dir is not None and options.agent is SkillAgentSelection.ALL:
        raise SkillError(
            "--skills-dir requires a single --agent",
            "custom skills directories are ambiguous when --agent all is used",
        )
    agent = options.agent
    scope = options.scope
    if agent is SkillAgentSelection.ALL and scope is SkillScope.PERSONAL:
        return [
            _resolve_target(spec, SkillAgent.CODEX, SkillScope.PERSONAL, None),
            _resolve_target(spec, SkillAgent.CLAUDE_CODE, SkillScope.PERSONAL, None),
            _resolve_target(spec, SkillAgent.OPENCODE, SkillScope.PERSONAL, None),
            _resolve_target(spec, SkillAgent.HERMES, SkillScope.PERSONAL, None),
        ]
    if agent is SkillAgentSelection.ALL and scope is SkillScope.WORKSPACE:
        return [
            _resolve_target(spec, SkillAgent.CODEX, SkillScope.WORKSPACE, None),
            _resolve_target(spec, SkillAgent.CLAUDE_CODE, SkillScope.WORKSPACE, None),
            _resolve_target(spec, SkillAgent.OPENCODE, SkillScope.WORKSPACE, None),
            _resolve_target(spec, SkillAgent.HERMES, SkillScope.WORKSPACE, None),
        ]
    if agent is SkillAgentSelection.CODEX:
        return [_resolve_target(spec, SkillAgent.CODEX, scope, options.skills_dir)]
    if agent is SkillAgentSelection.CLAUDE_CODE:
        return [_resolve_target(spec, SkillAgent.CLAUDE_CODE, scope, options.skills_dir)]
    if agent is SkillAgentSelection.OPENCODE:
        return [_resolve_target(spec, SkillAgent.OPENCODE, scope, options.skills_dir)]
    if agent is SkillAgentSelection.HERMES:
        return [_resolve_target(spec, SkillAgent.HERMES, scope, options.skills_dir)]
    raise SkillError(f"invalid --agent {agent!r}")


def _resolve_target(
    spec: SkillSpec,
    agent: SkillAgent,
    scope: SkillScope,
    skills_dir: Optional[str],
) -> _SkillTarget:
    base = _expand_tilde(skills_dir) if skills_dir is not None else _default_skills_dir(agent, scope)
    skill_dir = base / spec.name
    return _SkillTarget(
        agent=agent,
        scope=scope,
        skills_dir=base,
        skill_dir=skill_dir,
        skill_path=skill_dir / SKILL_FILE_NAME,
    )


def _default_skills_dir(agent: SkillAgent, scope: SkillScope) -> Path:
    if agent is SkillAgent.CODEX:
        if scope is not SkillScope.PERSONAL:
            return Path.cwd() / ".codex" / "skills"
        codex_home = os.environ.get("CODEX_HOME")
        if codex_home is not None:
            return Path(codex_home) / "skills"
        return _home_dir() / ".codex" / "skills"
    if agent is SkillAgent.CLAUDE_CODE:
        if scope is not SkillScope.PERSONAL:
            return Path.cwd() / ".claude" / "skills"
        return _home_dir() / ".claude" / "skills"
    if agent is SkillAgent.OPENCODE:
        if scope is not SkillScope.PERSONAL:
            return Path.cwd() / ".opencode" / "skills"
        xdg = os.environ.get("XDG_CONFIG_HOME")
        if xdg is not None:
            return Path(xdg) / "opencode" / "skills"
        return _home_dir() / ".config" / "opencode" / "skills"
    if agent is SkillAgent.HERMES:
        if scope is not SkillScope.PERSONAL:
            return Path.cwd() / ".hermes" / "skills"
        hermes_home = os.environ.get("HERMES_HOME")
        if hermes_home is not None:
            return Path(hermes_home) / "skills"
        return _home_dir() / ".hermes" / "skills"
    raise SkillError(f"unknown agent {agent!r}")


def _target_status(spec: SkillSpec, target: _SkillTarget) -> SkillTargetStatus:
    installed = target.skill_path.exists() or target.skill_path.is_symlink()
    valid = False
    current = False
    managed = False
    validation_error = None
    if target.skill_path.is_symlink():
        validation_error = "target SKILL.md is a symlink; refusing to follow it"
    elif target.skill_path.is_file():
        text = target.skill_path.read_text(encoding="utf-8")
        managed = _skill_text_is_managed_or_bundled(spec, text)
        current = not _skill_text_has_legacy_marker(spec, text) and _normalize_skill_text(spec, text) == _normalize_skill_text(spec, spec.source)
        try:
            _validate_skill_text(spec, text)
            valid = True
        except SkillError as exc:
            validation_error = exc.message
    elif installed:
        validation_error = "target SKILL.md is not a regular file"
    return SkillTargetStatus(
        agent=target.agent,
        scope=target.scope,
        skills_dir=str(target.skills_dir),
        skill_path=str(target.skill_path),
        installed=installed,
        managed=managed,
        valid=valid,
        current=current,
        validation_error=validation_error,
    )


def _target_uninstall_status(target: _SkillTarget, removed: bool) -> SkillUninstallStatus:
    return SkillUninstallStatus(
        agent=target.agent,
        scope=target.scope,
        skills_dir=str(target.skills_dir),
        skill_path=str(target.skill_path),
        removed=removed,
    )


def _generated_by(spec: SkillSpec) -> str:
    return f"Generated by {spec.marker_slug} skill install"


def _marker(spec: SkillSpec) -> str:
    return f"{spec.marker_slug}-managed-skill: true"


def _source_hash(spec: SkillSpec) -> str:
    h = _FNV1A64_OFFSET
    for byte in spec.source.encode("utf-8"):
        h ^= byte
        h = (h * _FNV1A64_PRIME) & _FNV1A64_MASK
    return f"{h:016x}"


def _managed_marker_block(spec: SkillSpec) -> str:
    slug = spec.marker_slug
    return (
        "<!--\n"
        f"{_generated_by(spec)}\n"
        f"{slug}-managed-skill: true\n"
        f"{slug}-managed-skill-name: {spec.name}\n"
        f"{slug}-managed-skill-source-hash-fnv1a64: {_source_hash(spec)}\n"
        "-->"
    )


def _legacy_marker_block(spec: SkillSpec) -> str:
    return f"<!-- {_generated_by(spec)} -->\n<!-- {_marker(spec)} -->"


def _managed_skill_contents(spec: SkillSpec) -> str:
    block = _managed_marker_block(spec)
    lines = _split_lines(spec.source)
    out: list[str] = []
    inserted = False
    for i, line in enumerate(lines):
        out.append(line)
        if i == 0:
            continue
        if not inserted and line.strip() == "---":
            out.append(block)
            out.append("")
            inserted = True
    if not inserted:
        out.append(block)
    return "".join(line + "\n" for line in out)


def _validate_installed_skill(spec: SkillSpec, path: Path) -> None:
    _validate_skill_text(spec, path.read_text(encoding="utf-8"))


def _validate_skill_text(spec: SkillSpec, text: str) -> None:
    frontmatter, error = _parse_skill_frontmatter(text)
    if error is not None:
        raise SkillError(
            f"invalid {spec.title} skill front matter: {error}",
            "quote scalar values that contain ': ', especially description",
        )
    assert frontmatter is not None
    if frontmatter.name != spec.name:
        raise SkillError(
            f"invalid {spec.title} skill front matter: name {frontmatter.name!r} does not match expected {spec.name!r}",
            f"set front matter name to {spec.name}",
        )


def _validate_skill_frontmatter(text: str) -> Optional[str]:
    _, error = _parse_skill_frontmatter(text)
    return error


@dataclass(frozen=True)
class _SkillFrontmatter:
    name: str


def _parse_skill_frontmatter(text: str) -> tuple[Optional[_SkillFrontmatter], Optional[str]]:
    lines = _split_lines(text)
    if not lines:
        return None, "missing YAML front matter"
    if lines[0].strip() != "---":
        return None, "missing opening --- YAML front matter delimiter"
    found_end = False
    name = None
    has_description = False
    for idx in range(1, len(lines)):
        line = lines[idx]
        line_no = idx + 1
        trimmed = line.strip()
        if trimmed == "---":
            found_end = True
            break
        if trimmed == "" or trimmed.startswith("#"):
            continue
        if line.startswith(" ") or line.startswith("\t"):
            return None, f"line {line_no}: nested YAML is not supported here"
        if ":" not in line:
            return None, f"line {line_no}: expected key: value"
        key, value = line.split(":", 1)
        key = key.strip()
        if key == "":
            return None, f"line {line_no}: empty key"
        value = value.lstrip()
        if key == "name":
            name = _parse_frontmatter_scalar(value)
        if key == "description":
            has_description = True
        if value.startswith('"') or value.startswith("'"):
            continue
        if ": " in value:
            return None, f"line {line_no}: unquoted scalar contains ': '; quote the value"
    if not found_end:
        return None, "missing closing --- YAML front matter delimiter"
    if name is None:
        return None, "missing required name field"
    if not has_description:
        return None, "missing required description field"
    return _SkillFrontmatter(name=name), None


def _parse_frontmatter_scalar(value: str) -> str:
    value = value.strip()
    if len(value) >= 2 and ((value[0] == '"' and value[-1] == '"') or (value[0] == "'" and value[-1] == "'")):
        return value[1:-1]
    return value


def _is_managed_or_bundled_skill(spec: SkillSpec, path: Path) -> bool:
    if not path.exists() and not path.is_symlink():
        return False
    if path.is_symlink():
        raise SkillError(
            f"refusing to inspect symlinked skill at {path}",
            "pass --force to replace or remove the symlink itself",
        )
    return _skill_text_is_managed_or_bundled(spec, path.read_text(encoding="utf-8"))


def _skill_text_is_managed_or_bundled(spec: SkillSpec, text: str) -> bool:
    if _skill_text_has_current_marker(spec, text) or _skill_text_has_legacy_marker(spec, text):
        return True
    return _normalize_skill_text(spec, text) == _normalize_skill_text(spec, spec.source)


def _skill_text_has_current_marker(spec: SkillSpec, text: str) -> bool:
    return _managed_marker_block(spec) in text.replace("\r\n", "\n")


def _skill_text_has_legacy_marker(spec: SkillSpec, text: str) -> bool:
    return _legacy_marker_block(spec) in text.replace("\r\n", "\n")


def _normalize_skill_text(spec: SkillSpec, text: str) -> str:
    """Drop the managed-marker blocks, then collapse runs of blank lines to one so
    that the blank line ``_managed_skill_contents`` inserts after the marker block
    does not make a managed install compare unequal to the bundled source.
    """
    text = text.replace("\r\n", "\n")
    text = text.replace(_managed_marker_block(spec), "")
    text = text.replace(_legacy_marker_block(spec), "")
    out: list[str] = []
    for line in _split_lines(text):
        trimmed = line.strip()
        if trimmed == "" and out and out[-1].strip() == "":
            continue
        out.append(line)
    return "\n".join(out).strip()


def _home_dir() -> Path:
    home = os.environ.get("HOME")
    if home is not None:
        return Path(home)
    home = os.environ.get("USERPROFILE")
    if home is not None:
        return Path(home)
    raise SkillError("cannot determine home directory", "pass --skills-dir explicitly")


def _expand_tilde(input_: str) -> Path:
    if input_ == "~":
        return _home_dir()
    if input_.startswith("~/"):
        return _home_dir() / input_[2:]
    return Path(input_)


def _validate_spec(spec: SkillSpec) -> None:
    _validate_slug("skill name", spec.name)
    _validate_slug("marker slug", spec.marker_slug)
    _validate_skill_text(spec, spec.source)


def _validate_slug(field: str, value: str) -> None:
    if _slug_is_valid(value):
        return
    raise SkillError(
        f"invalid {field} {value!r}: expected a lowercase slug matching [a-z0-9][a-z0-9-]*[a-z0-9]",
        "use lowercase ASCII letters, digits, and single hyphen-separated words",
    )


def _slug_is_valid(value: str) -> bool:
    if not value:
        return False
    def is_lower_alnum(ch: str) -> bool:
        return ("a" <= ch <= "z") or ("0" <= ch <= "9")
    return is_lower_alnum(value[0]) and is_lower_alnum(value[-1]) and all(is_lower_alnum(ch) or ch == "-" for ch in value)


def _write_skill_atomic(target: _SkillTarget, content: str) -> None:
    tmp_name = None
    try:
        fd, tmp_name = tempfile.mkstemp(prefix=f".{SKILL_FILE_NAME}.", suffix=".tmp", dir=target.skill_dir)
        with os.fdopen(fd, "w", encoding="utf-8") as tmp:
            tmp.write(content)
        os.chmod(tmp_name, 0o644)
        os.replace(tmp_name, target.skill_path)
        tmp_name = None
    except OSError as exc:
        raise SkillError(f"write skill failed: {exc}") from exc
    finally:
        if tmp_name is not None:
            try:
                os.unlink(tmp_name)
            except OSError:
                pass


def _split_lines(s: str) -> list[str]:
    """Mirror Rust's ``str::lines``: split on ``\\n`` (normalizing ``\\r\\n``),
    dropping a single trailing empty element from a final newline.
    """
    s = s.replace("\r\n", "\n")
    lines = s.split("\n")
    if lines and lines[-1] == "":
        lines.pop()
    return lines