EapilSkillMarket/scripts/validate_marketplace.py

#!/usr/bin/env python3
"""Validate this repository's Codex marketplace layout.

This is intentionally narrower than Codex's internal plugin validator. It checks
the structure this repo generates so sync automation can fail before
committing malformed marketplace data.
"""

from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
MARKETPLACES = [
    REPO_ROOT / ".agents" / "plugins" / "marketplace.json",
    REPO_ROOT / "plugins" / "codex" / ".agents" / "plugins" / "marketplace.json",
]
SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+([-.+][0-9A-Za-z.-]+)?$")
PLUGIN_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")


def load_json(path: Path) -> object:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except UnicodeDecodeError as exc:
        raise AssertionError(f"{path} is not valid UTF-8: {exc}") from exc
    except json.JSONDecodeError as exc:
        raise AssertionError(f"{path} is not valid JSON: {exc}") from exc


def require(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def resolve_manifest_path(plugin_dir: Path, manifest_path: Path, value: str, field: str) -> Path:
    require(value.startswith("./"), f"{manifest_path} {field} path must start with ./")
    target = (plugin_dir / value[2:]).resolve()
    plugin_root = plugin_dir.resolve()
    require(
        os.path.commonpath([str(plugin_root), str(target)]) == str(plugin_root),
        f"{manifest_path} {field} path must stay inside plugin root: {value}",
    )
    return target


def validate_manifest_path(plugin_dir: Path, manifest_path: Path, value: str, field: str) -> Path:
    target = resolve_manifest_path(plugin_dir, manifest_path, value, field)
    require(target.exists(), f"{manifest_path} {field} target does not exist: {target}")
    return target


def validate_hooks_field(plugin_dir: Path, manifest_path: Path, hooks: object) -> None:
    """Validate Codex-supported plugin hook declarations.

    Codex supports a hook path string, a list of path strings, an inline hooks
    object, or a list containing inline hooks objects. The marketplace only
    verifies path safety and basic type shape; Codex owns hook schema details.
    """

    if isinstance(hooks, str):
        validate_manifest_path(plugin_dir, manifest_path, hooks, "hooks")
        return
    if isinstance(hooks, dict):
        require(bool(hooks), f"{manifest_path} hooks inline object must not be empty")
        return
    if isinstance(hooks, list):
        require(bool(hooks), f"{manifest_path} hooks list must not be empty")
        for index, item in enumerate(hooks):
            if isinstance(item, str):
                validate_manifest_path(plugin_dir, manifest_path, item, f"hooks[{index}]")
            elif isinstance(item, dict):
                require(bool(item), f"{manifest_path} hooks[{index}] inline object must not be empty")
            else:
                raise AssertionError(f"{manifest_path} hooks[{index}] must be a path or inline object")
        return
    raise AssertionError(f"{manifest_path} hooks must be a path, path list, inline object, or inline object list")


def validate_manifest(plugin_dir: Path, expected_name: str) -> None:
    manifest_path = plugin_dir / ".codex-plugin" / "plugin.json"
    require(manifest_path.exists(), f"Missing manifest: {manifest_path}")
    manifest = load_json(manifest_path)
    require(isinstance(manifest, dict), f"Manifest must be an object: {manifest_path}")
    require(manifest.get("name") == expected_name, f"{manifest_path} name must be {expected_name}")
    require(PLUGIN_NAME_RE.match(expected_name) is not None, f"{manifest_path} plugin name must be ASCII kebab-case")
    require(SEMVER_RE.match(str(manifest.get("version", ""))) is not None, f"{manifest_path} version must be semver-like")
    require(bool(manifest.get("description")), f"{manifest_path} missing description")
    if "hooks" in manifest:
        validate_hooks_field(plugin_dir, manifest_path, manifest["hooks"])

    author = manifest.get("author")
    require(isinstance(author, dict) and bool(author.get("name")), f"{manifest_path} missing author.name")

    skills_rel = manifest.get("skills")
    require(isinstance(skills_rel, str) and skills_rel.startswith("./"), f"{manifest_path} skills path must be relative ./...")
    skills_dir = validate_manifest_path(plugin_dir, manifest_path, skills_rel, "skills")
    require(skills_dir.exists(), f"{manifest_path} skills directory does not exist: {skills_dir}")
    skill_dirs = [path for path in skills_dir.iterdir() if path.is_dir()]
    require(bool(skill_dirs), f"{manifest_path} must contain at least one skill directory")
    for skill_dir in skill_dirs:
        require((skill_dir / "SKILL.md").exists(), f"Missing SKILL.md in {skill_dir}")

    for optional_path_field in ["mcpServers", "apps"]:
        value = manifest.get(optional_path_field)
        if value is not None:
            require(isinstance(value, str), f"{manifest_path} {optional_path_field} must be a path string")
            validate_manifest_path(plugin_dir, manifest_path, value, optional_path_field)

    interface = manifest.get("interface")
    require(isinstance(interface, dict), f"{manifest_path} missing interface object")
    for key in ["displayName", "shortDescription", "longDescription", "developerName", "category"]:
        require(bool(interface.get(key)), f"{manifest_path} missing interface.{key}")
    capabilities = interface.get("capabilities")
    require(isinstance(capabilities, list), f"{manifest_path} interface.capabilities must be a list")


def resolve_marketplace_plugin_path(marketplace_path: Path, source_path: str) -> Path:
    require(source_path.startswith("./"), f"{marketplace_path} plugin source.path must start with ./")
    return (marketplace_path.parent.parent.parent / source_path[2:]).resolve()


def validate_marketplace(marketplace_path: Path) -> None:
    require(marketplace_path.exists(), f"Missing marketplace: {marketplace_path}")
    marketplace = load_json(marketplace_path)
    require(isinstance(marketplace, dict), f"Marketplace must be an object: {marketplace_path}")
    require(bool(marketplace.get("name")), f"{marketplace_path} missing name")
    require(isinstance(marketplace.get("plugins"), list), f"{marketplace_path} plugins must be a list")

    seen: set[str] = set()
    for entry in marketplace["plugins"]:
        require(isinstance(entry, dict), f"{marketplace_path} plugin entry must be an object")
        name = entry.get("name")
        require(isinstance(name, str) and name, f"{marketplace_path} plugin entry missing name")
        require(PLUGIN_NAME_RE.match(name) is not None, f"{marketplace_path} plugin name must be ASCII kebab-case: {name}")
        require(name not in seen, f"{marketplace_path} duplicate plugin name: {name}")
        seen.add(name)

        source = entry.get("source")
        require(isinstance(source, dict), f"{marketplace_path} {name} missing source")
        require(source.get("source") == "local", f"{marketplace_path} {name} source.source must be local")
        plugin_dir = resolve_marketplace_plugin_path(marketplace_path, str(source.get("path", "")))
        require(plugin_dir.exists(), f"{marketplace_path} {name} path does not exist: {plugin_dir}")

        policy = entry.get("policy")
        require(isinstance(policy, dict), f"{marketplace_path} {name} missing policy")
        require(policy.get("installation") in {"AVAILABLE", "INSTALLED_BY_DEFAULT", "NOT_AVAILABLE"}, f"{marketplace_path} {name} invalid installation policy")
        require(policy.get("authentication") in {"ON_INSTALL", "ON_USE"}, f"{marketplace_path} {name} invalid authentication policy")
        require(bool(entry.get("category")), f"{marketplace_path} {name} missing category")

        validate_manifest(plugin_dir, name)


def main() -> int:
    try:
        for marketplace in MARKETPLACES:
            validate_marketplace(marketplace)
    except AssertionError as exc:
        print(f"Validation failed: {exc}", file=sys.stderr)
        return 1
    print("Marketplace validation passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())