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, an empty no-op hooks object, or a list containing hook paths and
    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):
        return
    if isinstance(hooks, list):
        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):
                continue
            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 is_declared_component(manifest: dict[str, object], key: str) -> bool:
    if key not in manifest:
        return False
    value = manifest.get(key)
    if key != "hooks":
        return True
    if isinstance(value, str):
        return bool(value.strip())
    if isinstance(value, dict):
        return bool(value)
    if isinstance(value, list):
        return any(
            (isinstance(item, str) and bool(item.strip())) or (isinstance(item, dict) and bool(item))
            for item in value
        )
    return False


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 is_declared_component(manifest, key)]
    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())