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_skills_component(plugin_dir: Path, manifest_path: Path, skills_rel: object) -> None:
    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} skills component 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}")


def validate_json_component(
    plugin_dir: Path,
    manifest_path: Path,
    value: object,
    field: str,
    required_top_level: str,
) -> None:
    require(isinstance(value, str), f"{manifest_path} {field} must be a path string")
    target = validate_manifest_path(plugin_dir, manifest_path, value, field)
    payload = load_json(target)
    require(isinstance(payload, dict), f"{target} must be a JSON object")
    require(isinstance(payload.get(required_top_level), dict), f"{target} missing object {required_top_level}")


def validate_interface_asset(plugin_dir: Path, manifest_path: Path, value: object, field: str) -> None:
    if value is None or value == "":
        return
    require(isinstance(value, str), f"{manifest_path} interface.{field} must be a string")
    if value.startswith(("http://", "https://")):
        return
    validate_manifest_path(plugin_dir, manifest_path, value, f"interface.{field}")


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")
    author = manifest.get("author")
    require(isinstance(author, dict) and bool(author.get("name")), f"{manifest_path} missing author.name")

    declared_components = [
        key
        for key in ["skills", "mcpServers", "apps", "hooks"]
        if key in manifest
    ]
    require(bool(declared_components), f"{manifest_path} must declare at least one component: skills, mcpServers, apps, or hooks")

    if "skills" in manifest:
        validate_skills_component(plugin_dir, manifest_path, manifest.get("skills"))
    if "mcpServers" in manifest:
        validate_json_component(plugin_dir, manifest_path, manifest.get("mcpServers"), "mcpServers", "mcpServers")
    if "apps" in manifest:
        validate_json_component(plugin_dir, manifest_path, manifest.get("apps"), "apps", "apps")
    if "hooks" in manifest:
        validate_hooks_field(plugin_dir, manifest_path, manifest["hooks"])

    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")
    validate_interface_asset(plugin_dir, manifest_path, interface.get("logo"), "logo")
    validate_interface_asset(plugin_dir, manifest_path, interface.get("composerIcon"), "composerIcon")
    screenshots = interface.get("screenshots")
    if screenshots is not None:
        require(isinstance(screenshots, list), f"{manifest_path} interface.screenshots must be a list")
        for index, screenshot in enumerate(screenshots):
            validate_interface_asset(plugin_dir, manifest_path, screenshot, f"screenshots[{index}]")


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())