fix(review): harden plugin distribution contracts

scripts/generate_adapters.py

@@ -18,7 +18,16 @@ from typing import Any
 ROOT = Path(__file__).resolve().parents[1]
 CATALOG_PATH = Path("plugins/catalog.json")
 NAME_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
-HOST_ONLY_FRONTMATTER = {
+SEMVER_PATTERN = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$")
+FRONTMATTER_KEY_PATTERN = re.compile(r"^([a-z][a-z0-9-]*):(?:[ \t]*(.*))?$")
+PORTABLE_FRONTMATTER_FIELDS = {
+    "name",
+    "description",
+    "license",
+    "compatibility",
+    "metadata",
+}
+HOST_ONLY_FRONTMATTER_FIELDS = {
     "allowed-tools",
     "background",
     "model",
@@ -26,6 +35,28 @@ HOST_ONLY_FRONTMATTER = {
     "proactive",
     "tools",
 }
+ALLOWED_CAPABILITY_KINDS = {"skill", "agent", "hybrid"}
+ALLOWED_HOSTS = {"claude", "codex", "cursor", "opencode", "agent-skills"}
+NATIVE_HOSTS = {"claude", "codex", "cursor"}
+CLAUDE_OVERRIDE_FIELDS = {"agent_path", "description", "model"}
+CLAUDE_MODELS = {"inherit", "haiku", "sonnet", "opus"}
+REQUIRED_CATALOG_FIELDS = {
+    "schema_version",
+    "repository",
+    "defaults",
+    "marketplaces",
+    "plugins",
+}
+REQUIRED_PLUGIN_FIELDS = {
+    "id",
+    "version",
+    "description",
+    "package",
+    "capability",
+    "hosts",
+    "keywords",
+    "interface",
+}
 
 
 class GenerationFailure(RuntimeError):
@@ -44,31 +75,62 @@ def _parse_frontmatter(skill_file: Path, content: str) -> dict[str, str]:
     values: dict[str, str] = {}
     current_key: str | None = None
     for line in lines[1:end]:
-        if line.startswith((" ", "\t")) and current_key is not None:
-            values[current_key] = f"{values[current_key]} {line.strip()}".strip()
+        if not line.strip():
             continue
-        if ":" not in line:
+        if line.startswith((" ", "\t")):
+            if current_key not in {"description", "compatibility", "metadata"}:
+                raise GenerationFailure(
+                    f"{skill_file}: unexpected indented frontmatter content"
+                )
+            if current_key != "metadata":
+                values[current_key] = (
+                    f"{values[current_key]} {line.strip()}".strip()
+                )
             continue
-        key, value = line.split(":", 1)
-        current_key = key.strip()
-        values[current_key] = value.strip()
+        match = FRONTMATTER_KEY_PATTERN.fullmatch(line)
+        if match is None:
+            raise GenerationFailure(
+                f"{skill_file}: invalid frontmatter key syntax '{line}'"
+            )
+        current_key, value = match.groups()
+        if current_key in values:
+            raise GenerationFailure(
+                f"{skill_file}: duplicate frontmatter field '{current_key}'"
+            )
+        values[current_key] = (value or "").strip()
         if values[current_key] in {">", "|"}:
             values[current_key] = ""
     return values
 
 
 def _referenced_paths(content: str) -> set[str]:
-    references = set(re.findall(r"`([^`\n]+\.md)`", content))
-    references.update(re.findall(r"\]\(([^)\n]+)\)", content))
-    return {
-        reference.strip()
-        for reference in references
-        if not reference.startswith(("http://", "https://", "#"))
-        and (
+    inline_references = re.findall(
+        r"`([^`\n]+\.md(?:[?#][^`\n]*)?)`",
+        content,
+    )
+    markdown_references = re.findall(r"\]\(([^)\n]+)\)", content)
+    references: set[str] = set()
+    for raw_reference in inline_references:
+        reference = raw_reference.strip()
+        if not (
             "/" in reference
             or reference.startswith((".", "~"))
-        )
-    }
+        ):
+            continue
+        reference = reference.split("#", 1)[0].split("?", 1)[0].strip()
+        if reference:
+            references.add(reference)
+    for raw_reference in markdown_references:
+        reference = raw_reference.strip()
+        if (
+            reference.startswith(("#", "//"))
+            or re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", reference)
+        ):
+            continue
+        reference = reference.split("#", 1)[0].split("?", 1)[0].strip()
+        if reference:
+            references.add(reference)
+    return references
 
 
 def _case_mismatch(path: Path) -> bool:
@@ -125,8 +187,16 @@ def validate_skill(skill_directory: Path, content: str | None = None) -> list[st
     if not description or len(description) > 1024:
         errors.append(f"{skill_file}: description must contain 1-1024 characters")
 
-    for field in sorted(HOST_ONLY_FRONTMATTER & frontmatter.keys()):
-        errors.append(f"{skill_file}: host-only frontmatter '{field}' is not portable")
+    for field in sorted(frontmatter.keys() - PORTABLE_FRONTMATTER_FIELDS):
+        if field in HOST_ONLY_FRONTMATTER_FIELDS:
+            errors.append(
+                f"{skill_file}: host-only frontmatter '{field}' is not portable; "
+                f"non-portable frontmatter field '{field}'"
+            )
+        else:
+            errors.append(
+                f"{skill_file}: non-portable frontmatter field '{field}'"
+            )
 
     for reference in sorted(_referenced_paths(content)):
         pure_reference = PurePosixPath(reference)
@@ -168,9 +238,9 @@ def _render_claude_agent(plugin: dict[str, Any], content: str) -> str | None:
     model = claude.get("model", "inherit")
     return (
         "---\n"
-        f"name: {plugin['id']}\n"
-        f"description: {description}\n"
-        f"model: {model}\n"
+        f"name: {_yaml_scalar(plugin['id'])}\n"
+        f"description: {_yaml_scalar(description)}\n"
+        f"model: {_yaml_scalar(model)}\n"
         "---\n\n"
         "<!-- Generated from the canonical reviewer skill. Do not edit directly. -->\n\n"
         f"{_skill_body(content)}"
@@ -181,6 +251,290 @@ def _json_text(value: dict[str, Any]) -> str:
     return json.dumps(value, indent=2, ensure_ascii=False) + "\n"
 
 
+def _yaml_scalar(value: str) -> str:
+    if re.fullmatch(r"[A-Za-z][A-Za-z0-9 ./(),_+-]*", value):
+        return value
+    return json.dumps(value, ensure_ascii=False)
+
+
+def _require_string(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise GenerationFailure(f"{label} must be a non-empty string")
+    return value
+
+
+def _strict_catalog_path(
+    base: Path,
+    value: object,
+    *,
+    boundary: Path,
+    label: str,
+) -> Path:
+    if not isinstance(value, str) or not value:
+        raise GenerationFailure(f"{label} must be a non-empty relative path")
+    relative = PurePosixPath(value)
+    if (
+        relative.is_absolute()
+        or ".." in relative.parts
+        or "." in relative.parts
+        or not relative.parts
+    ):
+        raise GenerationFailure(f"{label} has unsafe path '{value}'")
+
+    base = base.absolute()
+    boundary = boundary.resolve()
+    candidate = base.joinpath(*relative.parts)
+    current = base
+    if current.is_symlink():
+        raise GenerationFailure(f"{label} has symlinked ancestor '{current}'")
+    for part in relative.parts:
+        current = current / part
+        if current.is_symlink():
+            raise GenerationFailure(f"{label} has symlinked ancestor '{current}'")
+
+    resolved = candidate.resolve(strict=False)
+    if resolved != boundary and boundary not in resolved.parents:
+        raise GenerationFailure(
+            f"{label} resolves outside '{boundary}': '{value}'"
+        )
+    return candidate
+
+
+def _validate_catalog(
+    catalog: dict[str, Any], root: Path
+) -> dict[str, dict[str, Path | None]]:
+    missing = sorted(REQUIRED_CATALOG_FIELDS - catalog.keys())
+    if missing:
+        raise GenerationFailure(
+            f"catalog is missing required field(s): {', '.join(missing)}"
+        )
+    if catalog.get("schema_version") != 1:
+        raise GenerationFailure("catalog schema_version must be 1")
+    _require_string(catalog.get("repository"), "catalog repository")
+
+    defaults = catalog.get("defaults")
+    if not isinstance(defaults, dict):
+        raise GenerationFailure("catalog defaults must be an object")
+    author = defaults.get("author")
+    if not isinstance(author, dict):
+        raise GenerationFailure("catalog defaults.author must be an object")
+    for field in ("name", "url"):
+        _require_string(
+            author.get(field),
+            f"catalog defaults.author.{field}",
+        )
+    for field in ("homepage", "license", "category"):
+        _require_string(defaults.get(field), f"catalog defaults.{field}")
+
+    marketplaces = catalog.get("marketplaces")
+    if not isinstance(marketplaces, dict):
+        raise GenerationFailure("catalog marketplaces must be an object")
+    for host in ("claude", "codex", "cursor"):
+        marketplace = marketplaces.get(host)
+        if not isinstance(marketplace, dict):
+            raise GenerationFailure(
+                f"catalog marketplaces.{host} must be an object"
+            )
+        for field in ("name", "display_name"):
+            _require_string(
+                marketplace.get(field),
+                f"catalog marketplaces.{host}.{field}",
+            )
+
+    plugins = catalog.get("plugins")
+    if not isinstance(plugins, list) or not plugins:
+        raise GenerationFailure("catalog plugins must be a non-empty array")
+
+    resolved_paths: dict[str, dict[str, Path | None]] = {}
+    for index, plugin in enumerate(plugins):
+        if not isinstance(plugin, dict):
+            raise GenerationFailure(f"catalog plugin at index {index} must be an object")
+        missing_plugin = sorted(REQUIRED_PLUGIN_FIELDS - plugin.keys())
+        if missing_plugin:
+            raise GenerationFailure(
+                f"catalog plugin at index {index} is missing field(s): "
+                f"{', '.join(missing_plugin)}"
+            )
+        plugin_id = _require_string(
+            plugin.get("id"), f"catalog plugin at index {index} id"
+        )
+        if not NAME_PATTERN.fullmatch(plugin_id):
+            raise GenerationFailure(f"plugin '{plugin_id}' has invalid id")
+        if plugin_id in resolved_paths:
+            raise GenerationFailure(f"duplicate plugin id '{plugin_id}'")
+        version = _require_string(
+            plugin.get("version"), f"plugin '{plugin_id}' version"
+        )
+        if not SEMVER_PATTERN.fullmatch(version):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' has invalid version '{version}'"
+            )
+        _require_string(
+            plugin.get("description"), f"plugin '{plugin_id}' description"
+        )
+
+        hosts = plugin.get("hosts")
+        if not isinstance(hosts, list) or not hosts:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' hosts must be a non-empty array"
+            )
+        if len(hosts) != len(set(host for host in hosts if isinstance(host, str))):
+            raise GenerationFailure(f"plugin '{plugin_id}' has duplicate hosts")
+        unsupported_hosts = [
+            host
+            for host in hosts
+            if not isinstance(host, str) or host not in ALLOWED_HOSTS
+        ]
+        if unsupported_hosts:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' has unsupported host "
+                f"'{unsupported_hosts[0]}'"
+            )
+        missing_portable_hosts = sorted(
+            {"opencode", "agent-skills"} - set(hosts)
+        )
+        if missing_portable_hosts:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' must support the canonical portable "
+                f"host(s): {', '.join(missing_portable_hosts)}"
+            )
+
+        package = _strict_catalog_path(
+            root,
+            plugin.get("package"),
+            boundary=root / "plugins",
+            label=f"plugin '{plugin_id}' package",
+        )
+        if package.parent != (root / "plugins").resolve():
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' package must be a direct child of 'plugins/'"
+            )
+        if not package.is_dir():
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' package does not exist: "
+                f"'{plugin.get('package')}'"
+            )
+
+        capability = plugin.get("capability")
+        if not isinstance(capability, dict):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' capability must be an object"
+            )
+        kind = capability.get("kind")
+        if kind not in ALLOWED_CAPABILITY_KINDS:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' has unsupported capability kind '{kind}'"
+            )
+        canonical_file = _strict_catalog_path(
+            root,
+            capability.get("canonical_path"),
+            boundary=root,
+            label=f"plugin '{plugin_id}' capability.canonical_path",
+        )
+        package_file = _strict_catalog_path(
+            package,
+            capability.get("package_path"),
+            boundary=package,
+            label=f"plugin '{plugin_id}' capability.package_path",
+        )
+        if canonical_file.name != "SKILL.md" or package_file.name != "SKILL.md":
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' skill paths must point to SKILL.md"
+            )
+        legacy_file = None
+        if capability.get("legacy_package_path") is not None:
+            legacy_file = _strict_catalog_path(
+                package,
+                capability.get("legacy_package_path"),
+                boundary=package,
+                label=f"plugin '{plugin_id}' capability.legacy_package_path",
+            )
+
+        overrides = plugin.get("overrides", {})
+        if not isinstance(overrides, dict):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' overrides must be an object"
+            )
+        claude = overrides.get("claude", {})
+        if not isinstance(claude, dict):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' Claude override must be an object"
+            )
+        unknown_override_fields = sorted(
+            set(claude) - CLAUDE_OVERRIDE_FIELDS
+        )
+        if unknown_override_fields:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' has unsupported Claude override "
+                f"field '{unknown_override_fields[0]}'"
+            )
+        agent_file = None
+        if "agent_path" in claude:
+            agent_file = _strict_catalog_path(
+                package,
+                claude.get("agent_path"),
+                boundary=package,
+                label=f"plugin '{plugin_id}' Claude agent_path",
+            )
+        if agent_file is not None and kind not in {"agent", "hybrid"}:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' Claude agent_path requires an "
+                "agent or hybrid capability"
+            )
+        if (
+            "claude" in hosts
+            and kind in {"agent", "hybrid"}
+            and agent_file is None
+        ):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' requires a Claude agent_path"
+            )
+        if "description" in claude:
+            _require_string(
+                claude.get("description"),
+                f"plugin '{plugin_id}' Claude description",
+            )
+        if "model" in claude and claude.get("model") not in CLAUDE_MODELS:
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' has invalid Claude model "
+                f"'{claude.get('model')}'"
+            )
+
+        interface = plugin.get("interface")
+        if not isinstance(interface, dict):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' interface must be an object"
+            )
+        for field in (
+            "display_name",
+            "short_description",
+            "long_description",
+        ):
+            _require_string(
+                interface.get(field),
+                f"plugin '{plugin_id}' interface.{field}",
+            )
+        prompts = interface.get("default_prompts")
+        if (
+            not isinstance(prompts, list)
+            or not prompts
+            or any(not isinstance(prompt, str) or not prompt for prompt in prompts)
+        ):
+            raise GenerationFailure(
+                f"plugin '{plugin_id}' interface.default_prompts "
+                "must be a non-empty string array"
+            )
+
+        resolved_paths[plugin_id] = {
+            "package": package,
+            "canonical_file": canonical_file,
+            "package_file": package_file,
+            "legacy_file": legacy_file,
+            "agent_file": agent_file,
+        }
+    return resolved_paths
+
+
 def _semver_key(version: str) -> tuple[int, int, int]:
     major, minor, patch = version.split(".")
     return int(major), int(minor), int(patch)
@@ -215,8 +569,6 @@ def _interface(plugin: dict[str, Any], defaults: dict[str, Any]) -> dict[str, An
 def _native_outputs(
     catalog: dict[str, Any], root: Path
 ) -> list[tuple[str, Path]]:
-    if not catalog.get("marketplaces") or not catalog.get("defaults"):
-        return []
     defaults = catalog["defaults"]
     outputs: list[tuple[str, Path]] = []
     claude_plugins: list[dict[str, Any]] = []
@@ -227,6 +579,7 @@ def _native_outputs(
     for plugin in catalog["plugins"]:
         versions.append(plugin["version"])
         package = root / plugin["package"]
+        hosts = set(plugin["hosts"])
         skill_root = (
             "./portable-skills/"
             if plugin["capability"]["kind"] == "hybrid"
@@ -260,51 +613,56 @@ def _native_outputs(
             "category": defaults["category"],
             "skills": skill_root,
         }
-        outputs.extend(
-            (
+        if "claude" in hosts:
+            outputs.append(
                 (
                     _json_text(claude_manifest),
                     package / ".claude-plugin" / "plugin.json",
-                ),
+                )
+            )
+            claude_plugins.append(
+                {
+                    "name": plugin["id"],
+                    "source": f"./{plugin['package']}",
+                    "description": plugin["description"],
+                    "category": "development",
+                }
+            )
+        if "codex" in hosts:
+            outputs.append(
                 (
                     _json_text(codex_manifest),
                     package / ".codex-plugin" / "plugin.json",
-                ),
+                )
+            )
+            codex_plugins.append(
+                {
+                    "name": plugin["id"],
+                    "source": {
+                        "source": "local",
+                        "path": f"./{plugin['package']}",
+                    },
+                    "policy": {
+                        "installation": "AVAILABLE",
+                        "authentication": "ON_INSTALL",
+                    },
+                    "category": defaults["category"],
+                }
+            )
+        if "cursor" in hosts:
+            outputs.append(
                 (
                     _json_text(cursor_manifest),
                     package / ".cursor-plugin" / "plugin.json",
-                ),
+                )
+            )
+            cursor_plugins.append(
+                {
+                    "name": plugin["id"],
+                    "source": f"./{plugin['package']}",
+                    "description": plugin["description"],
+                }
             )
-        )
-        claude_plugins.append(
-            {
-                "name": plugin["id"],
-                "source": f"./{plugin['package']}",
-                "description": plugin["description"],
-                "category": "development",
-            }
-        )
-        codex_plugins.append(
-            {
-                "name": plugin["id"],
-                "source": {
-                    "source": "local",
-                    "path": f"./{plugin['package']}",
-                },
-                "policy": {
-                    "installation": "AVAILABLE",
-                    "authentication": "ON_INSTALL",
-                },
-                "category": defaults["category"],
-            }
-        )
-        cursor_plugins.append(
-            {
-                "name": plugin["id"],
-                "source": f"./{plugin['package']}",
-                "description": plugin["description"],
-            }
-        )
 
     collection_version = max(versions, key=_semver_key)
     outputs.extend(
@@ -367,14 +725,26 @@ def _load_catalog(root: Path) -> dict[str, Any]:
     return value
 
 
-def _skill_projection(plugin: dict[str, Any], root: Path) -> tuple[Path, Path] | None:
-    capability = plugin.get("capability", {})
-    canonical_value = capability.get("canonical_path")
-    package_value = capability.get("package_path")
-    if not canonical_value or not package_value:
+def _needs_skill_projection(plugin: dict[str, Any]) -> bool:
+    native_hosts = set(plugin["hosts"]) & NATIVE_HOSTS
+    kind = plugin["capability"]["kind"]
+    if kind == "agent":
+        return False
+    if kind == "hybrid":
+        return bool(native_hosts & {"codex", "cursor"})
+    return bool(native_hosts)
+
+
+def _skill_projection(
+    plugin: dict[str, Any],
+    resolved: dict[str, Path | None],
+) -> tuple[Path, Path] | None:
+    if not _needs_skill_projection(plugin):
         return None
-    canonical_file = root / canonical_value
-    package_file = root / plugin["package"] / package_value
+    canonical_file = resolved["canonical_file"]
+    package_file = resolved["package_file"]
+    assert isinstance(canonical_file, Path)
+    assert isinstance(package_file, Path)
     if canonical_file.name != "SKILL.md" or package_file.name != "SKILL.md":
         raise GenerationFailure(
             f"plugin '{plugin.get('id')}' skill paths must point to SKILL.md"
@@ -411,12 +781,9 @@ def _replace_file(content: str, target: Path) -> None:
     os.replace(temporary, target)
 
 
-def _remove_legacy_projection(plugin: dict[str, Any], root: Path) -> None:
-    capability = plugin["capability"]
-    legacy_value = capability.get("legacy_package_path")
-    if not legacy_value:
+def _remove_legacy_projection(legacy_file: Path | None) -> None:
+    if legacy_file is None:
         return
-    legacy_file = root / plugin["package"] / legacy_value
     if legacy_file.is_file() or legacy_file.is_symlink():
         legacy_file.unlink()
     legacy_references = legacy_file.parent / "references"
@@ -425,37 +792,58 @@ def _remove_legacy_projection(plugin: dict[str, Any], root: Path) -> None:
 
 
 def _orphan_generated_paths(
-    catalog: dict[str, Any], root: Path
+    catalog: dict[str, Any],
+    root: Path,
+    resolved_paths: dict[str, dict[str, Path | None]],
 ) -> list[Path]:
     declared_packages = {
-        (root / plugin["package"]).resolve()
-        for plugin in catalog.get("plugins", [])
-        if isinstance(plugin, dict) and isinstance(plugin.get("package"), str)
+        resolved["package"]
+        for resolved in resolved_paths.values()
     }
     orphans: list[Path] = []
     for package in sorted((root / "plugins").iterdir()):
-        if not package.is_dir() or package.resolve() in declared_packages:
+        if (
+            not package.is_dir()
+            or package.is_symlink()
+            or package in declared_packages
+        ):
             continue
         manifests = [
             package / f".{target}-plugin" / "plugin.json"
             for target in ("claude", "codex", "cursor")
         ]
-        if not all(path.is_file() for path in manifests):
+        manifests = [path for path in manifests if path.is_file()]
+        if not manifests:
             continue
-        try:
-            names = {json.loads(path.read_text()).get("name") for path in manifests}
-        except json.JSONDecodeError:
-            continue
-        if names != {package.name}:
+        names: set[object] = set()
+        for path in manifests:
+            try:
+                manifest = json.loads(path.read_text())
+            except json.JSONDecodeError as error:
+                raise GenerationFailure(
+                    f"{path}: invalid JSON in generated orphan manifest "
+                    f"at line {error.lineno}, column {error.colno}"
+                ) from error
+            if not isinstance(manifest, dict):
+                raise GenerationFailure(
+                    f"{path}: generated orphan manifest must be an object"
+                )
+            names.add(manifest.get("name"))
+        if (
+            len(names) != 1
+            or not isinstance(next(iter(names)), str)
+            or not NAME_PATTERN.fullmatch(next(iter(names)))
+        ):
             continue
+        plugin_id = next(iter(names))
         orphans.extend(manifests)
         for skill in (
-            package / "skills" / package.name,
-            package / "portable-skills" / package.name,
+            package / "skills" / plugin_id,
+            package / "portable-skills" / plugin_id,
         ):
             if skill.is_dir() and not skill.is_symlink():
                 orphans.append(skill)
-        agent = package / "agents" / f"{package.name}.md"
+        agent = package / "agents" / f"{plugin_id}.md"
         if (
             agent.is_file()
             and "Generated from the canonical reviewer skill" in agent.read_text()
@@ -464,6 +852,39 @@ def _orphan_generated_paths(
     return orphans
 
 
+def _disabled_generated_paths(
+    catalog: dict[str, Any],
+    resolved_paths: dict[str, dict[str, Path | None]],
+) -> list[Path]:
+    disabled: list[Path] = []
+    for plugin in catalog["plugins"]:
+        resolved = resolved_paths[plugin["id"]]
+        package = resolved["package"]
+        assert isinstance(package, Path)
+        hosts = set(plugin["hosts"])
+        for host in sorted(NATIVE_HOSTS - hosts):
+            manifest = package / f".{host}-plugin" / "plugin.json"
+            if manifest.is_file():
+                disabled.append(manifest)
+
+        package_file = resolved["package_file"]
+        assert isinstance(package_file, Path)
+        package_skill = package_file.parent
+        if not _needs_skill_projection(plugin) and package_skill.is_dir():
+            disabled.append(package_skill)
+
+        agent_file = resolved["agent_file"]
+        if (
+            "claude" not in hosts
+            and isinstance(agent_file, Path)
+            and agent_file.is_file()
+            and "Generated from the canonical reviewer skill"
+            in agent_file.read_text()
+        ):
+            disabled.append(agent_file)
+    return disabled
+
+
 def _remove_generated_path(path: Path) -> None:
     parent = path.parent
     if path.is_dir() and not path.is_symlink():
@@ -474,54 +895,144 @@ def _remove_generated_path(path: Path) -> None:
         parent.rmdir()
 
 
+def _path_exists(path: Path) -> bool:
+    return path.exists() or path.is_symlink()
+
+
+def _remove_path(path: Path) -> None:
+    if path.is_dir() and not path.is_symlink():
+        shutil.rmtree(path)
+    elif _path_exists(path):
+        path.unlink()
+
+
+class _MutationJournal:
+    def __init__(self, root: Path, journal_root: Path, targets: list[Path]):
+        self.root = root
+        self.journal_root = journal_root
+        self.targets = self._without_nested_targets(targets)
+        self.backups: list[tuple[Path, Path | None]] = []
+        self.missing_parents: set[Path] = set()
+
+    @staticmethod
+    def _without_nested_targets(targets: list[Path]) -> list[Path]:
+        unique = sorted(set(targets), key=lambda path: (len(path.parts), str(path)))
+        selected: list[Path] = []
+        for path in unique:
+            if any(parent == path or parent in path.parents for parent in selected):
+                continue
+            selected.append(path)
+        return selected
+
+    def snapshot(self) -> None:
+        backup_root = self.journal_root / "rollback"
+        backup_root.mkdir()
+        for index, target in enumerate(self.targets):
+            parent = target.parent
+            while parent != self.root and self.root in parent.parents:
+                if not parent.exists():
+                    self.missing_parents.add(parent)
+                parent = parent.parent
+            if not _path_exists(target):
+                self.backups.append((target, None))
+                continue
+            backup = backup_root / str(index)
+            if target.is_dir() and not target.is_symlink():
+                shutil.copytree(target, backup)
+            elif target.is_symlink():
+                backup.symlink_to(os.readlink(target))
+            else:
+                backup.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copy2(target, backup)
+            self.backups.append((target, backup))
+
+    def rollback(self) -> None:
+        for target, _ in sorted(
+            self.backups,
+            key=lambda item: len(item[0].parts),
+            reverse=True,
+        ):
+            _remove_path(target)
+        for target, backup in self.backups:
+            if backup is None:
+                continue
+            target.parent.mkdir(parents=True, exist_ok=True)
+            if backup.is_dir() and not backup.is_symlink():
+                shutil.copytree(backup, target)
+            elif backup.is_symlink():
+                target.symlink_to(os.readlink(backup))
+            else:
+                shutil.copy2(backup, target)
+        for parent in sorted(
+            self.missing_parents,
+            key=lambda path: len(path.parts),
+            reverse=True,
+        ):
+            if parent.is_dir() and not any(parent.iterdir()):
+                parent.rmdir()
+
+
 def generate_adapters(root: Path = ROOT, check: bool = False) -> list[str]:
     root = root.resolve()
     catalog = _load_catalog(root)
+    resolved_paths = _validate_catalog(catalog, root)
     projections: list[tuple[dict[str, Any], Path, Path]] = []
     generated_files: list[tuple[str, Path]] = []
     validation_errors: list[str] = []
-    for plugin in catalog.get("plugins", []):
-        if not isinstance(plugin, dict):
-            continue
-        projection = _skill_projection(plugin, root)
-        if projection is None:
-            continue
-        source, target = projection
-        source_content = (source / "SKILL.md").read_text()
-        errors = validate_skill(source, source_content)
-        validation_errors.extend(errors)
-        projections.append((plugin, source, target))
-        agent_content = _render_claude_agent(plugin, source_content)
-        agent_path = plugin.get("overrides", {}).get("claude", {}).get("agent_path")
-        if agent_content is not None and agent_path:
+    for plugin in catalog["plugins"]:
+        resolved = resolved_paths[plugin["id"]]
+        canonical_file = resolved["canonical_file"]
+        assert isinstance(canonical_file, Path)
+        source = canonical_file.parent
+        source_content = canonical_file.read_text()
+        validation_errors.extend(validate_skill(source, source_content))
+
+        projection = _skill_projection(plugin, resolved)
+        if projection is not None:
+            _, target = projection
+            projections.append((plugin, source, target))
+
+        agent_file = resolved["agent_file"]
+        if "claude" in plugin["hosts"] and isinstance(agent_file, Path):
+            agent_content = _render_claude_agent(plugin, source_content)
+            assert agent_content is not None
             generated_files.append(
                 (
                     agent_content,
-                    root / plugin["package"] / agent_path,
+                    agent_file,
                 )
             )
     if validation_errors:
         raise GenerationFailure("\n".join(validation_errors))
     generated_files.extend(_native_outputs(catalog, root))
-    orphan_paths = _orphan_generated_paths(catalog, root)
+    orphan_paths = _orphan_generated_paths(catalog, root, resolved_paths)
+    disabled_paths = _disabled_generated_paths(catalog, resolved_paths)
 
     changed: list[str] = []
     for _, source, target in projections:
         if _tree_files(source) != _tree_files(target):
             changed.append(str(target.relative_to(root)))
-    for plugin, _, _ in projections:
-        legacy = plugin["capability"].get("legacy_package_path")
-        if legacy and (root / plugin["package"] / legacy).exists():
-            target_directory = root / plugin["package"] / Path(
-                plugin["capability"]["package_path"]
-            ).parent
-            relative_target = str(target_directory.relative_to(root))
+    legacy_paths: list[Path] = []
+    for plugin in catalog["plugins"]:
+        resolved = resolved_paths[plugin["id"]]
+        legacy_file = resolved["legacy_file"]
+        if isinstance(legacy_file, Path) and (
+            _path_exists(legacy_file)
+            or (legacy_file.parent / "references").is_dir()
+        ):
+            legacy_paths.extend(
+                [legacy_file, legacy_file.parent / "references"]
+            )
+            package_file = resolved["package_file"]
+            assert isinstance(package_file, Path)
+            relative_target = str(package_file.parent.relative_to(root))
             if relative_target not in changed:
                 changed.append(relative_target)
     for content, target in generated_files:
         if not target.is_file() or target.read_text() != content:
             changed.append(str(target.relative_to(root)))
-    changed.extend(str(path.relative_to(root)) for path in orphan_paths)
+    deletion_paths = sorted(set(orphan_paths + disabled_paths))
+    changed.extend(str(path.relative_to(root)) for path in deletion_paths)
 
     changed = sorted(set(changed))
     if check or not changed:
@@ -537,15 +1048,37 @@ def generate_adapters(root: Path = ROOT, check: bool = False) -> list[str]:
             shutil.copytree(source, stage)
             staged.append((plugin, stage, target))
 
-        for _, stage, target in staged:
-            _replace_directory(stage, target)
-        for plugin, _, _ in projections:
-            _remove_legacy_projection(plugin, root)
-        for path in orphan_paths:
-            _remove_generated_path(path)
-        for content, target in generated_files:
-            if str(target.relative_to(root)) in changed:
-                _replace_file(content, target)
+        changed_files = [
+            target
+            for _, target in generated_files
+            if str(target.relative_to(root)) in changed
+        ]
+        mutation_targets = (
+            [target for _, _, target in staged]
+            + legacy_paths
+            + deletion_paths
+            + changed_files
+        )
+        journal = _MutationJournal(root, stage_root, mutation_targets)
+        journal.snapshot()
+        try:
+            for _, stage, target in staged:
+                _replace_directory(stage, target)
+            for plugin in catalog["plugins"]:
+                legacy_file = resolved_paths[plugin["id"]]["legacy_file"]
+                _remove_legacy_projection(
+                    legacy_file if isinstance(legacy_file, Path) else None
+                )
+            for path in deletion_paths:
+                _remove_generated_path(path)
+            for content, target in generated_files:
+                if str(target.relative_to(root)) in changed:
+                    _replace_file(content, target)
+        except Exception as error:
+            journal.rollback()
+            raise GenerationFailure(
+                f"adapter generation transaction failed: {error}"
+            ) from error
     return changed
 
 

scripts/smoke_plugins.py

@@ -15,6 +15,7 @@ from typing import Any, Iterable
 
 
 ROOT = Path(__file__).resolve().parents[1]
+OPENCODE_TIMEOUT_SECONDS = 10
 
 
 class SmokeFailure(RuntimeError):
@@ -67,10 +68,13 @@ def discover_capabilities(package: Path, target: str) -> list[dict[str, Any]]:
     return capabilities
 
 
-def install_package(root: Path, home: Path, plugin_id: str) -> Path:
-    source = root / "plugins" / plugin_id
+def install_package(root: Path, home: Path, plugin: dict[str, Any]) -> Path:
+    plugin_id = plugin["id"]
+    source = root / plugin["package"]
     if not source.is_dir():
-        raise SmokeFailure(f"unknown plugin '{plugin_id}'")
+        raise SmokeFailure(
+            f"plugin '{plugin_id}' has no package source at '{plugin['package']}'"
+        )
     plugins_home = home / "plugins"
     plugins_home.mkdir(parents=True, exist_ok=True)
     destination = plugins_home / plugin_id
@@ -90,18 +94,32 @@ def find_duplicates(
     registrations: Iterable[tuple[str, str]],
 ) -> dict[str, dict[str, Any]]:
     provenance: dict[str, list[str]] = defaultdict(list)
-    for plugin_id, source in registrations:
+    for plugin_id, source in sorted(registrations):
         provenance[plugin_id].append(source)
     return {
         plugin_id: {
             "provenance": sources,
             "remediation": "keep one installation channel and remove the shadow copy",
         }
-        for plugin_id, sources in provenance.items()
+        for plugin_id, sources in sorted(provenance.items())
         if len(sources) > 1
     }
 
 
+def _raise_for_duplicates(registrations: Iterable[tuple[str, str]]) -> None:
+    duplicates = find_duplicates(registrations)
+    if not duplicates:
+        return
+    reports = [
+        (
+            f"{plugin_id}: provenance={','.join(details['provenance'])}; "
+            f"remediation={details['remediation']}"
+        )
+        for plugin_id, details in duplicates.items()
+    ]
+    raise SmokeFailure(f"duplicate registrations detected: {'; '.join(reports)}")
+
+
 def _smoke_opencode(
     root: Path, home: Path, catalog: dict[str, Any]
 ) -> list[dict[str, Any]]:
@@ -120,12 +138,19 @@ def _smoke_opencode(
         "hooks: Object.keys(hooks), paths: config.skills.paths"
         "}));"
     )
-    completed = subprocess.run(
-        ["node", "--input-type=module", "--eval", script],
-        check=False,
-        capture_output=True,
-        text=True,
-    )
+    try:
+        completed = subprocess.run(
+            ["node", "--input-type=module", "--eval", script],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=OPENCODE_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired as error:
+        raise SmokeFailure(
+            f"OpenCode adapter '{adapter}' timed out after "
+            f"{OPENCODE_TIMEOUT_SECONDS} seconds"
+        ) from error
     if completed.returncode:
         raise SmokeFailure(
             f"OpenCode adapter failed to load: {completed.stderr.strip()}"
@@ -148,6 +173,7 @@ def _smoke_opencode(
             "provenance": "opencode-plugin",
         }
         for plugin in catalog["plugins"]
+        if "opencode" in plugin["hosts"]
     ]
 
 
@@ -156,7 +182,11 @@ def _smoke_agent_skills(
 ) -> list[dict[str, Any]]:
     skills_home = home / "skills"
     shutil.copytree(root / "skills", skills_home)
-    expected = sorted(plugin["id"] for plugin in catalog["plugins"])
+    expected = sorted(
+        plugin["id"]
+        for plugin in catalog["plugins"]
+        if "agent-skills" in plugin["hosts"]
+    )
     discovered = sorted(path.parent.name for path in skills_home.glob("*/SKILL.md"))
     if discovered != expected:
         raise SmokeFailure(
@@ -182,17 +212,27 @@ def _smoke_in_home(
     if target == "agent-skills":
         return _smoke_agent_skills(root, home, catalog)
     results: list[dict[str, Any]] = []
+    registrations: list[tuple[str, str]] = []
     for plugin in catalog["plugins"]:
-        installed = install_package(root, home, plugin["id"])
+        if target not in plugin["hosts"]:
+            continue
+        installed = install_package(root, home, plugin)
         capabilities = discover_capabilities(installed, target)
+        provenance = f"{target}-marketplace"
         results.append(
             {
                 "plugin": plugin["id"],
                 "status": "ok",
                 "capability": capabilities[0]["id"],
-                "provenance": f"{target}-marketplace",
+                "provenance": provenance,
             }
         )
+        registrations.append((capabilities[0]["id"], provenance))
+    registrations.extend(
+        (skill["id"], "direct-agent-skill")
+        for skill in _discover_skills(home / "skills")
+    )
+    _raise_for_duplicates(registrations)
     return results
 
 

scripts/validate_plugins.py

@@ -28,8 +28,18 @@ REQUIRED_PLUGIN_FIELDS = {
     "keywords",
     "interface",
 }
+REQUIRED_CATALOG_FIELDS = {
+    "schema_version",
+    "repository",
+    "defaults",
+    "marketplaces",
+    "plugins",
+}
 ALLOWED_CAPABILITY_KINDS = {"skill", "agent", "hybrid"}
 ALLOWED_HOSTS = {"claude", "codex", "cursor", "opencode", "agent-skills"}
+NATIVE_HOSTS = {"claude", "codex", "cursor"}
+CLAUDE_OVERRIDE_FIELDS = {"agent_path", "description", "model"}
+CLAUDE_MODELS = {"inherit", "haiku", "sonnet", "opus"}
 SEMVER = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$")
 
 
@@ -60,6 +70,49 @@ def _safe_relative_path(value: object) -> PurePosixPath | None:
     return path
 
 
+def _resolve_catalog_path(
+    base: Path,
+    value: object,
+    *,
+    boundary: Path,
+) -> tuple[Path | None, str | None]:
+    relative = _safe_relative_path(value)
+    if relative is None:
+        return None, f"unsafe path '{value}'"
+    base = base.absolute()
+    boundary = boundary.resolve()
+    candidate = base.joinpath(*relative.parts)
+    current = base
+    if current.is_symlink():
+        return None, f"symlinked ancestor '{current}'"
+    for part in relative.parts:
+        current = current / part
+        if current.is_symlink():
+            return None, f"symlinked ancestor '{current}'"
+    resolved = candidate.resolve(strict=False)
+    if resolved != boundary and boundary not in resolved.parents:
+        return None, f"path '{value}' resolves outside '{boundary}'"
+    return candidate, None
+
+
+def _needs_skill_projection(plugin: dict[str, Any]) -> bool:
+    hosts = plugin.get("hosts")
+    if not isinstance(hosts, list):
+        return False
+    native_hosts = set(
+        host for host in hosts if isinstance(host, str)
+    ) & NATIVE_HOSTS
+    capability = plugin.get("capability")
+    if not isinstance(capability, dict):
+        return False
+    kind = capability.get("kind")
+    if kind == "agent":
+        return False
+    if kind == "hybrid":
+        return bool(native_hosts & {"codex", "cursor"})
+    return bool(native_hosts)
+
+
 def _plugin_label(plugin: object, index: int) -> str:
     if isinstance(plugin, dict) and isinstance(plugin.get("id"), str):
         return f"plugin '{plugin['id']}'"
@@ -68,8 +121,57 @@ def _plugin_label(plugin: object, index: int) -> str:
 
 def validate_catalog(catalog: dict[str, Any], root: Path = ROOT) -> list[str]:
     errors: list[str] = []
+    for field in sorted(REQUIRED_CATALOG_FIELDS - catalog.keys()):
+        errors.append(f"catalog is missing required field '{field}'")
     if catalog.get("schema_version") != 1:
         errors.append("catalog schema_version must be 1")
+    if not isinstance(catalog.get("repository"), str) or not catalog[
+        "repository"
+    ].strip():
+        errors.append("catalog repository must be a non-empty string")
+    defaults = catalog.get("defaults")
+    if not isinstance(defaults, dict):
+        errors.append("catalog defaults must be an object")
+    else:
+        author = defaults.get("author")
+        if not isinstance(author, dict):
+            errors.append("catalog defaults.author must be an object")
+        else:
+            for field in ("name", "url"):
+                if (
+                    not isinstance(author.get(field), str)
+                    or not author[field].strip()
+                ):
+                    errors.append(
+                        f"catalog defaults.author.{field} must be "
+                        "a non-empty string"
+                    )
+        for field in ("homepage", "license", "category"):
+            if (
+                not isinstance(defaults.get(field), str)
+                or not defaults[field].strip()
+            ):
+                errors.append(
+                    f"catalog defaults.{field} must be a non-empty string"
+                )
+    marketplaces = catalog.get("marketplaces")
+    if not isinstance(marketplaces, dict):
+        errors.append("catalog marketplaces must be an object")
+    else:
+        for host in ("claude", "codex", "cursor"):
+            marketplace = marketplaces.get(host)
+            if not isinstance(marketplace, dict):
+                errors.append(f"catalog marketplaces.{host} must be an object")
+                continue
+            for field in ("name", "display_name"):
+                if (
+                    not isinstance(marketplace.get(field), str)
+                    or not marketplace[field].strip()
+                ):
+                    errors.append(
+                        f"catalog marketplaces.{host}.{field} must be "
+                        "a non-empty string"
+                    )
 
     plugins = catalog.get("plugins")
     if not isinstance(plugins, list) or not plugins:
@@ -103,16 +205,24 @@ def validate_catalog(catalog: dict[str, Any], root: Path = ROOT) -> list[str]:
             errors.append(f"{label} has invalid description")
 
         package_value = plugin.get("package")
-        package_relative = _safe_relative_path(package_value)
         package_root: Path | None = None
         package_exists = False
-        if package_relative is None:
-            errors.append(f"{label} has unsafe package path '{package_value}'")
+        package_root, package_error = _resolve_catalog_path(
+            root,
+            package_value,
+            boundary=root / "plugins",
+        )
+        if package_error is not None:
+            errors.append(f"{label} package has {package_error}")
         else:
-            package_root = root.joinpath(*package_relative.parts)
+            assert package_root is not None
             package_exists = package_root.is_dir()
             if not package_exists:
                 errors.append(f"{label} references unknown package root '{package_value}'")
+            if package_root.parent != (root / "plugins").resolve():
+                errors.append(
+                    f"{label} package must be a direct child of 'plugins/'"
+                )
 
         capability = plugin.get("capability")
         if not isinstance(capability, dict):
@@ -121,17 +231,48 @@ def validate_catalog(catalog: dict[str, Any], root: Path = ROOT) -> list[str]:
             kind = capability.get("kind")
             if kind not in ALLOWED_CAPABILITY_KINDS:
                 errors.append(f"{label} has unsupported capability kind '{kind}'")
+            canonical_value = capability.get("canonical_path")
+            _, canonical_error = _resolve_catalog_path(
+                root,
+                canonical_value,
+                boundary=root,
+            )
+            if canonical_error is not None:
+                errors.append(
+                    f"{label} canonical capability path has {canonical_error}"
+                )
             surface_value = capability.get("package_path")
-            surface_relative = _safe_relative_path(surface_value)
-            if surface_relative is None:
-                errors.append(f"{label} has unsafe capability path '{surface_value}'")
-            elif package_root is not None and package_exists:
-                surface = package_root.joinpath(*surface_relative.parts)
-                if surface.is_symlink() or not surface.is_file():
+            surface: Path | None = None
+            if package_root is not None:
+                surface, surface_error = _resolve_catalog_path(
+                    package_root,
+                    surface_value,
+                    boundary=package_root,
+                )
+            else:
+                surface_error = f"unsafe path '{surface_value}'"
+            if surface_error is not None:
+                errors.append(
+                    f"{label} package capability path has {surface_error}"
+                )
+            elif package_exists and _needs_skill_projection(plugin):
+                assert surface is not None
+                if not surface.is_file():
                     errors.append(
                         f"{label} advertises no usable surface at "
                         f"'{package_value}/{surface_value}'"
                     )
+            legacy_value = capability.get("legacy_package_path")
+            if legacy_value is not None and package_root is not None:
+                _, legacy_error = _resolve_catalog_path(
+                    package_root,
+                    legacy_value,
+                    boundary=package_root,
+                )
+                if legacy_error is not None:
+                    errors.append(
+                        f"{label} legacy capability path has {legacy_error}"
+                    )
 
         hosts = plugin.get("hosts")
         if not isinstance(hosts, list) or not hosts:
@@ -144,6 +285,43 @@ def validate_catalog(catalog: dict[str, Any], root: Path = ROOT) -> list[str]:
                 errors.append(f"{label} has unsupported host '{host}'")
             if len(hosts) != len(set(host for host in hosts if isinstance(host, str))):
                 errors.append(f"{label} has duplicate hosts")
+            missing_portable_hosts = sorted(
+                {"opencode", "agent-skills"} - set(hosts)
+            )
+            if missing_portable_hosts:
+                errors.append(
+                    f"{label} canonical skills require host(s): "
+                    f"{', '.join(missing_portable_hosts)}"
+                )
+
+        overrides = plugin.get("overrides", {})
+        if not isinstance(overrides, dict):
+            errors.append(f"{label} overrides must be an object")
+        else:
+            claude = overrides.get("claude", {})
+            if not isinstance(claude, dict):
+                errors.append(f"{label} Claude override must be an object")
+            else:
+                unknown_fields = sorted(
+                    set(claude) - CLAUDE_OVERRIDE_FIELDS
+                )
+                for field in unknown_fields:
+                    errors.append(
+                        f"{label} has unsupported Claude override field '{field}'"
+                    )
+                model = claude.get("model")
+                if model is not None and model not in CLAUDE_MODELS:
+                    errors.append(f"{label} has invalid Claude model '{model}'")
+                if "agent_path" in claude and package_root is not None:
+                    _, agent_error = _resolve_catalog_path(
+                        package_root,
+                        claude.get("agent_path"),
+                        boundary=package_root,
+                    )
+                    if agent_error is not None:
+                        errors.append(
+                            f"{label} Claude agent path has {agent_error}"
+                        )
 
     return errors
 
@@ -161,7 +339,11 @@ def validate_marketplace(
     catalog_plugins = {
         plugin["id"]: plugin
         for plugin in catalog.get("plugins", [])
-        if isinstance(plugin, dict) and isinstance(plugin.get("id"), str)
+        if (
+            isinstance(plugin, dict)
+            and isinstance(plugin.get("id"), str)
+            and target in plugin.get("hosts", [])
+        )
     }
     entries = marketplace.get("plugins")
     if not isinstance(entries, list):