package_internal_plugin.py

#!/usr/bin/env python3
"""Build a company-branded appsec-advisor plugin artifact.

This script keeps internal packaging logic in the upstream plugin instead of
copying fragile rsync/sed/json snippets into every company's CI repository.
"""

from __future__ import annotations

import argparse
import gzip
import hashlib
import io
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import tarfile
import tempfile
from datetime import datetime, timezone
from pathlib import Path

UPSTREAM_NAMESPACE = "appsec-advisor"
# Namespace rewriting + leak detection must reach the script surfaces too:
# scripts/run-headless.sh and many .py helpers hardcode `<namespace>:<skill>`
# command references (e.g. the `claude -p` prompt, `/…:fix-run-issues` hints).
# Excluding .sh/.py left a repackaged plugin dispatching the upstream namespace
# — a broken headless wrapper that check_namespace_leaks never saw.
TEXT_SUFFIXES = {".json", ".md", ".txt", ".yaml", ".yml", ".sh", ".py"}
TOP_LEVEL_EXCLUDES = {
    ".agents",
    ".cache",
    ".claude",
    ".codex",
    ".env",
    ".git",
    ".github",
    ".gitlab-ci.yml",
    ".gitignore",
    ".pytest_cache",
    ".ruff_cache",
    ".mypy_cache",
    ".venv",
    ".venv-tests",
    "AGENTS.md",
    "CHANGELOG.md",
    "CLAUDE.md",
    "CONTRIBUTING.md",
    "LICENSE",
    "Makefile",
    "README.md",
    "SECURITY.md",
    "build",
    "dist",
    "examples",
    "htmlcov",
    "node_modules",
    "pyproject.toml",
    "tests",
}
ANY_LEVEL_EXCLUDES = {
    ".appsec-cache",
    ".cache",
    ".coverage-data",
    ".mypy_cache",
    ".pytest_cache",
    ".ruff_cache",
    ".venv",
    ".venv-tests",
    "__pycache__",
    "build",
    "dist",
    "htmlcov",
    "node_modules",
}
ANY_LEVEL_FILE_EXCLUDES = {
    ".agent-run.log",
    ".appsec-checkpoint",
    ".appsec-lock",
    ".appsec-progress.json",
    ".hook-events.log",
    ".render-integrity.json",
    ".run-issues.json",
    ".scan-start-epoch",
    ".scan-wall-seconds",
    ".section-integrity.json",
    ".skill-config.json",
}
PATH_EXCLUDES = {
    ("data", "appsec-requirements-fallback.yaml"),
    ("docs", "analysis"),
    ("docs", "internal"),
    ("docs", "proposals"),
    ("docs", "security"),
    ("scripts", "docs"),
    ("skills", "appsec-analyzer", "docs"),
    ("tests", "fixtures", "e2e", "_last-run"),
}
NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
SURFACE_MANIFEST = ".claude-plugin/package-surface.json"
MCP_CONFIG = ".mcp.json"
HOOK_SCRIPT_IDS = {
    "agent_logger.py": "agent-logger",
    "security_steering.py": "security-coach",
}
INTERNAL_REQUIRED_SKILLS = {"internal-threat-analysis-kernel"}


def _die(message: str, code: int = 2) -> None:
    print(f"ERROR: {message}", file=sys.stderr)
    raise SystemExit(code)


def _validate_package_name(name: str) -> None:
    if not NAME_RE.match(name):
        _die(
            "plugin name must start with a lowercase letter or digit and contain "
            "only lowercase letters, digits, '.', '_' and '-'"
        )


def _validate_version(version: str) -> None:
    if not version:
        _die("version must not be empty")
    if "/" in version:
        _die("VERSION must not contain '/' because it is used in artifact paths")


def _require_plugin_root(source: Path) -> None:
    required = [
        ".claude-plugin/plugin.json",
        "config.json",
        "agents",
        "skills",
        "scripts",
        "schemas",
    ]
    missing = [rel for rel in required if not (source / rel).exists()]
    if missing:
        _die(f"{source} is not an appsec-advisor plugin root; missing {missing}")


def _copy_ignore(source_root: Path):
    source_root = source_root.resolve()

    def ignore(current: str, names: list[str]) -> set[str]:
        current_path = Path(current).resolve()
        try:
            rel = current_path.relative_to(source_root)
        except ValueError:
            rel = Path(".")

        ignored: set[str] = set()
        for name in names:
            child = current_path / name
            rel_child = tuple((rel / name).parts)
            if rel == Path(".") and name in TOP_LEVEL_EXCLUDES:
                ignored.add(name)
            elif rel_child in PATH_EXCLUDES:
                ignored.add(name)
            elif child.is_dir() and name in ANY_LEVEL_EXCLUDES:
                ignored.add(name)
            elif child.is_file() and name in ANY_LEVEL_FILE_EXCLUDES:
                ignored.add(name)
        return ignored

    return ignore


def copy_source(source: Path, build: Path) -> None:
    if build.exists():
        shutil.rmtree(build)
    build.parent.mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, build, ignore=_copy_ignore(source))


def overlay_org_profile(org_profile: Path, build: Path) -> None:
    if not (org_profile / "org-profile.yaml").is_file():
        _die(f"{org_profile} must contain org-profile.yaml")
    target = build / "org-profile"
    if target.exists():
        shutil.rmtree(target)
    shutil.copytree(org_profile, target)


def _git(source: Path, *cmd: str) -> str:
    """Read-only git output from the source checkout, or empty when unavailable."""
    try:
        result = subprocess.run(["git", *cmd], capture_output=True, text=True, timeout=10, cwd=source, check=False)
    except (OSError, subprocess.SubprocessError):
        return ""
    return result.stdout.strip() if result.returncode == 0 else ""


def core_provenance(source: Path, version: str) -> dict:
    """Which upstream revision this package was built from.

    The build tree carries no ``.git``, so a packaged install can no longer read
    its own origin: without these fields ``status`` shows the organization's
    version number and nothing about the appsec-advisor underneath it. Every
    field is best-effort — a source tree that is not a git checkout records the
    core version alone rather than failing the build.
    """
    provenance = {"appsec_advisor_core_version": version}
    ref = _git(source, "rev-parse", "--abbrev-ref", "HEAD")
    if ref == "HEAD":  # detached: name the tag when one points here
        ref = _git(source, "describe", "--tags", "--exact-match") or "detached"
    commit = _git(source, "rev-parse", "HEAD")
    committed_at = _git(source, "log", "-1", "--format=%cI")
    if ref:
        provenance["appsec_advisor_core_ref"] = ref
    if commit:
        provenance["appsec_advisor_core_commit"] = commit
    if committed_at:
        provenance["appsec_advisor_core_committed_at"] = committed_at
    if _git(source, "status", "--porcelain"):
        provenance["appsec_advisor_core_dirty"] = True
    provenance["appsec_advisor_packaged_at"] = (
        datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    )
    return provenance


def upstream_url_from_git(source: Path) -> str | None:
    """The source checkout's origin remote as an https URL, when it has one.

    Used when the caller passed no ``--upstream-url``: the packaged build is the
    only place a reader can still see where the code came from, and an origin
    remote answers that as well as the flag does. An SSH remote is rewritten to
    its https form; anything else is dropped rather than guessed at.
    """
    remote = _git(source, "remote", "get-url", "origin")
    if not remote:
        return None
    match = re.match(r"^(?:git\+)?ssh://git@([^/]+)/(.+?)(?:\.git)?/?$", remote) or re.match(
        r"^git@([^:]+):(.+?)(?:\.git)?/?$", remote
    )
    if match:
        return f"https://{match.group(1)}/{match.group(2)}"
    if remote.startswith("https://"):
        return remote.removesuffix(".git")
    return None


def patch_plugin_json(
    build: Path, name: str, version: str, description: str | None, source: Path, *, provenance: dict | None = None
) -> None:
    plugin_path = build / ".claude-plugin" / "plugin.json"
    data = json.loads(plugin_path.read_text(encoding="utf-8"))
    core_version = data.get("version", "")
    data["name"] = name
    data["version"] = version
    data["description"] = (
        description if description is not None else f"Internal packaged build of appsec-advisor for {name}."
    )
    data.update(core_provenance(source, core_version) if provenance is None else provenance)
    plugin_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


def _org_profile_banner(build: Path) -> dict:
    """Session-banner overrides declared in the packaged org profile.

    The banner runs as a SessionStart hook and must stay free of a YAML
    dependency, so the profile is resolved into config.json here at build time.
    """
    profile_path = build / "org-profile" / "org-profile.yaml"
    if not profile_path.is_file():
        return {}
    banner = _load_yaml_or_json(profile_path).get("banner")
    if not isinstance(banner, dict):
        return {}
    return {key: value for key, value in banner.items() if key in ("enabled", "headline", "url")}


def _org_profile_baseline(build: Path) -> dict:
    """The organization's own secure-coding baseline, resolved into config.json.

    Like the banner, this is read at build time so the SessionStart hook stays
    free of a YAML dependency. Declaring any source *replaces* the plugin's
    default baseline rather than merging with it: the upstream source and the
    upstream bundled copy both carry the upstream id, which the organization's
    own id check would refuse anyway, so keeping them would only produce a
    confusing failure. ``file`` is rewritten to the packaged profile path,
    because the profile directory is where it lands in the build.
    """
    profile_path = build / "org-profile" / "org-profile.yaml"
    if not profile_path.is_file():
        return {}
    baseline = _load_yaml_or_json(profile_path).get("baseline")
    if not isinstance(baseline, dict):
        return {}

    resolved = {
        key: value for key, value in baseline.items() if key in ("enabled", "enforce", "id", "name", "install_filename")
    }
    if baseline.get("url"):
        resolved["url"] = baseline["url"]
    if isinstance(baseline.get("git"), dict):
        resolved["git"] = baseline["git"]
    if baseline.get("file"):
        resolved["fallback_file"] = f"org-profile/{str(baseline['file']).lstrip('/')}"

    if any(key in resolved for key in ("id", "url", "git", "fallback_file")):
        # An organization that names its own baseline owns the whole chain; a
        # leftover upstream source would be fetched, fail the id check, and
        # report a puzzling error. The display name goes the same way: inherited,
        # it would head the banner and both skills with the upstream product
        # name over the organization's own rules. Cleared, the generic default
        # applies until the profile sets one.
        for key in ("url", "git", "release", "fallback_file", "name", "bundle_dir"):
            resolved.setdefault(key, None)
        resolved["mode"] = "complete"
    return resolved


def _org_profile_skill_toggles(build: Path) -> dict:
    """The profile's skill policy, resolved into config.json at build time.

    Without this the policy only exists in ``.org-profile-effective.json``,
    which a run writes into its output directory — so a skill invoked before
    any scan has ever happened found no policy and ran. Resolving it here gives
    every session the same answer from the first command onwards, the way the
    banner and baseline blocks already work.

    Normalised to the ``{enabled, reason}`` shape so consumers see one form,
    matching ``resolve_org_profile.normalize_skill_toggles``.
    """
    profile_path = build / "org-profile" / "org-profile.yaml"
    if not profile_path.is_file():
        return {}
    toggles = _load_yaml_or_json(profile_path).get("skill_toggles")
    if not isinstance(toggles, dict):
        return {}
    out: dict[str, dict] = {}
    for name, value in toggles.items():
        if isinstance(value, bool):
            out[str(name)] = {"enabled": value, "reason": None}
        elif isinstance(value, dict):
            out[str(name)] = {"enabled": bool(value.get("enabled", True)), "reason": value.get("reason")}
    return out


def _prune_unused_baselines(build: Path, baseline: dict) -> None:
    """Drop the upstream bundled baseline when the build does not use it.

    An organization that ships its own baseline never reads the copy in
    ``data/baselines/`` — its id would not match, so an install would refuse it.
    Leaving it in place ships a third-party document, under its own licence,
    that the build's configuration never touches: dead weight in an internal
    package and a puzzle for whoever audits it.
    """
    bundled_dir = build / "data" / "baselines"
    if not bundled_dir.is_dir():
        return
    fallback = baseline.get("fallback_file")
    if isinstance(fallback, str) and fallback.startswith("data/baselines/"):
        return
    shutil.rmtree(bundled_dir, ignore_errors=True)


def patch_config(build: Path, info_url: str | None = None) -> None:
    config_path = build / "config.json"
    data = json.loads(config_path.read_text(encoding="utf-8"))
    data["organization_profile"] = {
        "enabled": True,
        "path": "org-profile/org-profile.yaml",
    }
    baseline = dict(data.get("baseline") or {})
    baseline.update(_org_profile_baseline(build))
    if baseline:
        data["baseline"] = baseline
        _prune_unused_baselines(build, baseline)
    toggles = _org_profile_skill_toggles(build)
    if toggles:
        data["skill_toggles"] = toggles
    banner = dict(data.get("banner") or {})
    banner.update(_org_profile_banner(build))
    if info_url is not None:
        # The flag wins over the profile. An empty value drops the "more
        # information" line; the upstream URL is wrong for an internal build
        # either way.
        banner["url"] = info_url or None
    if banner:
        data["banner"] = banner
    config_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


def _load_yaml_or_json(path: Path) -> dict:
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        _die(f"cannot read package policy {path}: {exc}")
    if path.suffix.lower() == ".json":
        try:
            data = json.loads(text)
        except json.JSONDecodeError as exc:
            _die(f"invalid JSON package policy {path}: {exc}")
    else:
        try:
            import yaml
        except ImportError:
            _die("package policy YAML requires PyYAML; install pyyaml or use a .json policy file")
        try:
            data = yaml.safe_load(text)
        except yaml.YAMLError as exc:
            _die(f"invalid YAML package policy {path}: {exc}")
    if data is None:
        return {}
    if not isinstance(data, dict):
        _die(f"package policy {path} must contain a mapping/object at the root")
    return data


def load_package_policy(org_profile: Path, explicit_path: str | None) -> tuple[dict, Path | None]:
    if explicit_path:
        path = Path(explicit_path).resolve()
        if not path.is_file():
            _die(f"package policy not found at {path}")
        return _load_yaml_or_json(path), path

    for name in ("package-policy.yaml", "package-policy.yml", "package-policy.json"):
        candidate = org_profile / name
        if candidate.is_file():
            return _load_yaml_or_json(candidate), candidate.resolve()
    return {}, None


def _policy_surface(policy: dict) -> dict:
    surface = policy.get("plugin_surface", policy)
    if not isinstance(surface, dict):
        _die("package policy 'plugin_surface' must be a mapping/object")
    unknown = set(surface) - {"skills", "hooks", "mcp_servers"}
    if unknown:
        _die(f"package policy has unknown plugin_surface keys: {sorted(unknown)}")
    return surface


def _read_name_list(block: dict, key: str, surface: str) -> set[str] | None:
    if key not in block:
        return None
    value = block[key]
    if value is None:
        return set()
    if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
        _die(f"package policy plugin_surface.{surface}.{key} must be a list of strings")
    seen: set[str] = set()
    duplicates: set[str] = set()
    for item in value:
        name = item.strip()
        if not name:
            _die(f"package policy plugin_surface.{surface}.{key} contains an empty name")
        if name in seen:
            duplicates.add(name)
        seen.add(name)
    if duplicates:
        _die(f"package policy plugin_surface.{surface}.{key} contains duplicates: {sorted(duplicates)}")
    return seen


def _resolve_keep_set(
    block: object,
    available: set[str],
    surface: str,
    *,
    required: set[str] | None = None,
) -> set[str]:
    required = required or set()
    if block is None:
        return set(available)
    if not isinstance(block, dict):
        _die(f"package policy plugin_surface.{surface} must be a mapping/object")
    unknown_keys = set(block) - {"include", "exclude"}
    if unknown_keys:
        _die(f"package policy plugin_surface.{surface} has unknown keys: {sorted(unknown_keys)}")
    include = _read_name_list(block, "include", surface)
    exclude = _read_name_list(block, "exclude", surface)
    if include is not None and exclude is not None:
        _die(f"package policy plugin_surface.{surface} cannot set both include and exclude")

    selected = include if include is not None else exclude
    if selected is None:
        return set(available)
    unknown = selected - available
    if unknown:
        _die(
            f"package policy plugin_surface.{surface} references unknown names: "
            f"{sorted(unknown)} (available: {sorted(available)})"
        )
    keep = set(selected) if include is not None else (available - selected)
    missing_required = required - keep
    if missing_required:
        _die(f"package policy plugin_surface.{surface} must keep required names: {sorted(missing_required)}")
    return keep


def _available_skills(build: Path) -> set[str]:
    skills_dir = build / "skills"
    if not skills_dir.is_dir():
        return set()
    return {path.parent.name for path in skills_dir.glob("*/SKILL.md") if path.is_file()}


def _required_internal_skills(build: Path) -> set[str]:
    """Internal skills referenced by agent frontmatter in this packaged source."""
    agents_dir = build / "agents"
    if not agents_dir.is_dir():
        return set()
    required: set[str] = set()
    for path in agents_dir.glob("*.md"):
        try:
            text = path.read_text(encoding="utf-8")
        except OSError:
            continue
        frontmatter = text.split("---", 2)[1] if text.startswith("---") and text.count("---") >= 2 else ""
        for skill in INTERNAL_REQUIRED_SKILLS:
            if re.search(rf"(?m)^\s*-\s+{re.escape(skill)}\s*$", frontmatter):
                required.add(skill)
    return required


def _hook_id(command: str) -> str | None:
    if "/scripts/" not in command and "\\scripts\\" not in command:
        return None
    script_name = command.replace("\\", "/").split("/scripts/", 1)[1].split()[0]
    script_name = Path(script_name).name
    return HOOK_SCRIPT_IDS.get(script_name, Path(script_name).stem.replace("_", "-"))


def _load_hooks(build: Path) -> tuple[Path, dict]:
    hooks_path = build / "hooks" / "hooks.json"
    if not hooks_path.is_file():
        return hooks_path, {"hooks": {}}
    try:
        data = json.loads(hooks_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        _die(f"invalid hooks.json in packaged copy: {exc}")
    if not isinstance(data, dict) or not isinstance(data.get("hooks"), dict):
        _die("hooks/hooks.json must contain a top-level 'hooks' object")
    return hooks_path, data


def _available_hook_ids(build: Path) -> set[str]:
    _, data = _load_hooks(build)
    ids: set[str] = set()
    for entries in data.get("hooks", {}).values():
        if not isinstance(entries, list):
            continue
        for outer in entries:
            if not isinstance(outer, dict):
                continue
            for hook in outer.get("hooks") or []:
                if not isinstance(hook, dict):
                    continue
                command = hook.get("command")
                if isinstance(command, str):
                    hook_id = _hook_id(command)
                    if hook_id:
                        ids.add(hook_id)
    return ids


def apply_skill_policy(build: Path, surface: dict, *, required: set[str] | None = None) -> dict:
    available = _available_skills(build)
    required_internal = _required_internal_skills(build)
    missing_internal = required_internal - available
    if missing_internal:
        _die(f"packaged source is missing internal required skills: {sorted(missing_internal)}")
    keep = _resolve_keep_set(
        surface.get("skills"),
        available,
        "skills",
        required={"appsec-analyzer"} if required is None else required,
    )
    keep.update(required_internal)
    removed = sorted(available - keep)
    for skill in removed:
        shutil.rmtree(build / "skills" / skill)
    return {"included": sorted(keep), "removed": removed}


# Where an org profile keeps the skills it adds, unless it says otherwise.
ORG_SKILLS_GLOB = "skills/*/SKILL.md"


def overlay_org_skills(build: Path) -> list[str]:
    """Copy the skills the org profile adds into the plugin's skills directory.

    Claude Code discovers skills by convention — every ``skills/<name>/SKILL.md``
    under the plugin root — so adding one is a directory copy. It has to happen
    before the package policy runs, or ``plugin_surface.skills`` could not name
    an added skill and would reject it as unknown.

    Two refusals rather than a silent result:

    * A name that collides with an upstream skill aborts the build. Overwriting
      ``appsec-analyzer`` with an organization's own file would replace the
      pipeline's entry point without anyone deciding to.
    * Frontmatter that would not pass for an upstream skill aborts too. The
      description is the only text the model sees when choosing a skill, so a
      malformed one degrades routing quietly instead of failing.
    """
    profile_dir = build / "org-profile"
    profile_path = profile_dir / "org-profile.yaml"
    if not profile_path.is_file():
        return []
    block = _load_yaml_or_json(profile_path).get("skills")
    if not isinstance(block, dict):
        return []
    pattern = str(block.get("add") or ORG_SKILLS_GLOB)

    upstream = _available_skills(build)
    added: list[str] = []
    for skill_md in sorted(profile_dir.glob(pattern)):
        source = skill_md.parent
        name = source.name
        _require_inside(profile_dir, skill_md, "skills.add")
        _validate_org_skill(skill_md, name)
        if name in upstream:
            _die(
                f"org profile adds skill '{name}', which the plugin already ships. "
                f"Rename it — replacing an upstream skill would change what its command does."
            )
        if name in added:
            _die(f"org profile adds skill '{name}' more than once")
        shutil.copytree(source, build / "skills" / name)
        added.append(name)
    return added


def _require_inside(root: Path, path: Path, label: str) -> None:
    """Abort when ``path`` escapes ``root`` — a glob must not reach outside."""
    try:
        path.resolve().relative_to(root.resolve())
    except ValueError:
        _die(f"package policy {label}: '{path}' resolves outside the profile directory")


def _validate_org_skill(skill_md: Path, directory: str) -> None:
    """Hold an added skill to the same frontmatter rules as an upstream one."""
    lines = skill_md.read_text(encoding="utf-8").splitlines()
    if not lines or lines[0].strip() != "---":
        _die(f"org skill '{directory}': SKILL.md does not start with a '---' frontmatter delimiter")
    try:
        end = lines.index("---", 1)
    except ValueError:
        _die(f"org skill '{directory}': frontmatter is not terminated by a closing '---'")
    try:
        import yaml

        front = yaml.safe_load("\n".join(lines[1:end]))
    except Exception as exc:  # noqa: BLE001
        _die(f"org skill '{directory}': frontmatter is not valid YAML: {exc}")
    if not isinstance(front, dict):
        _die(f"org skill '{directory}': frontmatter must be a mapping")
    unknown = set(front) - {"name", "description"}
    if unknown:
        _die(f"org skill '{directory}': unsupported frontmatter keys {sorted(unknown)}")
    name, description = front.get("name"), front.get("description")
    if name != directory:
        _die(f"org skill '{directory}': frontmatter name '{name}' must match the directory name")
    if not isinstance(description, str) or not description.strip():
        _die(f"org skill '{directory}': a non-empty description is required")
    if len(description) > 1024:
        _die(f"org skill '{directory}': description is {len(description)} characters; the limit is 1024")


def _org_profile_hooks(build: Path) -> dict:
    """Hook definitions declared in the packaged org profile (top-level `hooks`).

    Returns a {id: {event, command, matcher?}} mapping, or {} when the profile
    has none. The org profile is already overlaid into build/org-profile here."""
    profile_path = build / "org-profile" / "org-profile.yaml"
    if not profile_path.is_file():
        return {}
    data = _load_yaml_or_json(profile_path)
    hooks = data.get("hooks")
    return hooks if isinstance(hooks, dict) else {}


def apply_hook_policy(build: Path, surface: dict) -> dict:
    org_hooks = _org_profile_hooks(build)
    # Org hook ids join the upstream ids in the keep-set resolution, so
    # plugin_surface.hooks (include/exclude) gates org hooks too. Declared org
    # hooks are included by default.
    upstream_available = _available_hook_ids(build)
    available = upstream_available | set(org_hooks)
    keep = _resolve_keep_set(surface.get("hooks"), available, "hooks")
    removed = sorted(available - keep)

    hooks_path, data = _load_hooks(build)
    filtered_events: dict[str, list[dict]] = {}
    for event, entries in data.get("hooks", {}).items():
        if not isinstance(entries, list):
            continue
        kept_entries: list[dict] = []
        for outer in entries:
            if not isinstance(outer, dict):
                continue
            hooks = outer.get("hooks") or []
            kept_hooks = []
            for hook in hooks:
                if not isinstance(hook, dict):
                    continue
                command = hook.get("command")
                hook_id = _hook_id(command) if isinstance(command, str) else None
                if hook_id is None or hook_id in keep:
                    kept_hooks.append(hook)
            if kept_hooks:
                new_outer = dict(outer)
                new_outer["hooks"] = kept_hooks
                kept_entries.append(new_outer)
        if kept_entries:
            filtered_events[event] = kept_entries

    # Merge kept org hooks into the built hooks.json under their declared event.
    org_kept: list[dict] = []
    for hook_id in sorted(keep):
        cfg = org_hooks.get(hook_id)
        if not isinstance(cfg, dict):
            continue
        event = cfg.get("event")
        command = cfg.get("command")
        if not event or not command:
            continue
        outer: dict = {"hooks": [{"type": "command", "command": command}]}
        if cfg.get("matcher"):
            outer["matcher"] = cfg["matcher"]
        filtered_events.setdefault(event, []).append(outer)
        org_kept.append({"id": hook_id, "event": event, "command": command})

    if hooks_path.parent.exists():
        hooks_path.write_text(
            json.dumps({"hooks": filtered_events}, indent=2) + "\n",
            encoding="utf-8",
        )

    if "security-coach" in removed:
        keywords_path = build / "hooks" / "steering_keywords.json"
        if keywords_path.exists():
            keywords_path.unlink()

    return {
        # Upstream hooks only — org hooks are tracked separately under "org"
        # (they are not discoverable via the /scripts/ id derivation).
        "included": sorted(upstream_available & keep),
        "removed": removed,
        "events": sorted(filtered_events),
        "org": org_kept,
    }


def _org_profile_mcp_servers(build: Path) -> dict:
    """MCP server definitions declared in the packaged org profile (mcp.servers).

    Returns a {name: server_config} mapping, or {} when the profile has no mcp
    block. The org profile has already been overlaid into build/org-profile at
    this point.
    """
    profile_path = build / "org-profile" / "org-profile.yaml"
    if not profile_path.is_file():
        return {}
    data = _load_yaml_or_json(profile_path)
    servers = (data.get("mcp") or {}).get("servers")
    return servers if isinstance(servers, dict) else {}


def apply_mcp_policy(build: Path, surface: dict) -> dict:
    """Emit build/.mcp.json from the org profile's mcp.servers, gated by the
    package-policy allowlist (plugin_surface.mcp_servers). Declared servers are
    included by default; an include/exclude list narrows them. Writes no file
    when nothing is kept."""
    servers = _org_profile_mcp_servers(build)
    available = set(servers)
    keep = _resolve_keep_set(surface.get("mcp_servers"), available, "mcp_servers")
    removed = sorted(available - keep)

    mcp_path = build / MCP_CONFIG
    if keep:
        payload = {"mcpServers": {name: servers[name] for name in sorted(keep)}}
        mcp_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    elif mcp_path.exists():
        mcp_path.unlink()
    return {"included": sorted(keep), "removed": removed}


def write_surface_manifest(
    build: Path,
    policy_path: Path | None,
    skills: dict,
    hooks: dict,
    upstream_url: str | None = None,
    mcp_servers: dict | None = None,
) -> None:
    if policy_path is None:
        policy_ref = None
    elif (build / "org-profile" / policy_path.name).is_file():
        policy_ref = f"org-profile/{policy_path.name}"
    else:
        try:
            policy_ref = str(policy_path.relative_to(build))
        except ValueError:
            policy_ref = policy_path.name
    manifest = {
        "version": 1,
        "policy": policy_ref,
        "skills": skills,
        "hooks": hooks,
    }
    if mcp_servers is not None:
        manifest["mcp_servers"] = mcp_servers
    if upstream_url:
        manifest["upstream_url"] = upstream_url
        manifest["based_on"] = upstream_url.removesuffix(".git")
    manifest_path = build / SURFACE_MANIFEST
    manifest_path.parent.mkdir(parents=True, exist_ok=True)
    manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")


def apply_package_surface_policy(
    build: Path,
    policy: dict,
    policy_path: Path | None,
    upstream_url: str | None = None,
    *,
    required_skills: set[str] | None = None,
) -> None:
    surface = _policy_surface(policy)
    # Added first, so the package policy can name an org skill like any other
    # and so the manifest counts it among what this build ships.
    org_added = overlay_org_skills(build)
    skills = apply_skill_policy(build, surface, required=required_skills)
    if org_added:
        # Recorded separately, the way org hooks are: the artifact surface has
        # to say which parts are the organization's own.
        skills["org_added"] = sorted(org_added)
    hooks = apply_hook_policy(build, surface)
    mcp_servers = apply_mcp_policy(build, surface)
    write_surface_manifest(build, policy_path, skills, hooks, upstream_url, mcp_servers)


def _has_shebang(path: Path) -> bool:
    try:
        with path.open("rb") as fh:
            return fh.read(2) == b"#!"
    except OSError:
        return False


def _text_files(root: Path):
    for path in root.rglob("*"):
        if not path.is_file():
            continue
        if path.suffix.lower() in TEXT_SUFFIXES:
            yield path
        elif not path.suffix and _has_shebang(path):
            # Extensionless executables (CLI shims like scripts/appsec-reviewer-cli)
            # hardcode `<namespace>:<skill>` command references too, but match no
            # suffix. Detect them by shebang so the namespace rewrite and the leak
            # check reach them — without hardcoding shim names.
            yield path


def rewrite_namespace(build: Path, name: str, *, protected: set[str] | None = None) -> None:
    old = f"{UPSTREAM_NAMESPACE}:"
    new = f"{name}:"
    for path in _text_files(build):
        if protected and path.relative_to(build).as_posix() in protected:
            continue
        try:
            text = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
        if old in text:
            path.write_text(text.replace(old, new), encoding="utf-8")


def check_namespace_leaks(build: Path, *, protected: set[str] | None = None) -> None:
    needle = f"{UPSTREAM_NAMESPACE}:"
    leaks: list[str] = []
    for root_name in ("skills", "agents", "scripts"):
        root = build / root_name
        if not root.exists():
            continue
        for path in _text_files(root):
            if protected and path.relative_to(build).as_posix() in protected:
                continue
            try:
                text = path.read_text(encoding="utf-8")
            except UnicodeDecodeError:
                continue
            if needle in text:
                leaks.append(str(path.relative_to(build)))
    if leaks:
        shown = "\n  - ".join(leaks[:20])
        _die(f"upstream namespace {needle!r} still present in packaged copy:\n  - {shown}", 1)


def run_validation(build: Path) -> None:
    subprocess.run(
        [sys.executable, str(build / "scripts" / "validators/validate_config.py"), str(build)],
        check=True,
    )
    subprocess.run(
        [
            sys.executable,
            str(build / "scripts" / "validators/validate_org_profile.py"),
            str(build / "org-profile" / "org-profile.yaml"),
        ],
        check=True,
    )


def write_archive(
    build: Path, name: str, version: str, dist_dir: Path, *, deterministic: bool = False
) -> tuple[Path, Path]:
    dist_dir.mkdir(parents=True, exist_ok=True)
    tar_path = dist_dir / f"{name}-{version}.tgz"
    sha_path = dist_dir / f"{name}-{version}.tgz.sha256"

    if deterministic:
        with (
            tar_path.open("wb") as stream,
            gzip.GzipFile(filename="", fileobj=stream, mode="wb", mtime=0) as compressed,
        ):
            with tarfile.open(fileobj=compressed, mode="w") as archive:
                for relative, data in sorted(_staging_tree(build).items()):
                    entry = tarfile.TarInfo(f"{name}/{relative}")
                    entry.size = len(data)
                    entry.mode = 0o644
                    archive.addfile(entry, io.BytesIO(data))
    else:
        with tarfile.open(tar_path, "w:gz") as archive:
            archive.add(build, arcname=name)

    digest = hashlib.sha256(tar_path.read_bytes()).hexdigest()
    sha_path.write_text(f"{digest}  {tar_path.name}\n", encoding="utf-8")
    return tar_path, sha_path


def remove_stale_archive(name: str, version: str, dist_dir: Path) -> None:
    for path in (
        dist_dir / f"{name}-{version}.tgz",
        dist_dir / f"{name}-{version}.tgz.sha256",
    ):
        if path.exists():
            path.unlink()


HOOK_DESCRIPTIONS = {
    "agent-logger": "Logs all tool calls and agent actions for audit and debugging.",
    "security-coach": "Intercepts tool calls and provides real-time security guidance.",
    "session-banner": "Opens each session with the repository's threat-model and baseline status.",
    "skill-policy-gate": "Blocks skills this organization disabled, on both the typed and the model-invoked path.",
}

# Skills with their own detailed section in the README
MAIN_SKILLS = ["appsec-analyzer", "audit-security-requirements", "verify-requirements"]
INTERNAL_SKILLS = INTERNAL_REQUIRED_SKILLS
# Skills grouped into a single utility section
UTILITY_SKILLS = [
    "help",
    "security-assessment-health",
    "check-permissions",
    "status",
    "fix-run-issues",
    "diagnose-run",
    "clean-run-state",
]


def _skill_description(build: Path, skill: str) -> str:
    """First sentence of a skill's description, for the packaged README table.

    The frontmatter is parsed as YAML rather than scanned line by line. Several
    skills write the description as a block scalar::

        description: >-
          Install the secure-coding baseline into ...

    where the value is on the following lines — a line scan returned the literal
    ``>-`` and the README table shipped that as the skill's description.
    """
    skill_md = build / "skills" / skill / "SKILL.md"
    if not skill_md.exists():
        return ""
    lines = skill_md.read_text(encoding="utf-8").splitlines()
    if not lines or lines[0].strip() != "---":
        return ""
    try:
        end = lines.index("---", 1)
    except ValueError:
        return ""
    try:
        import yaml

        front = yaml.safe_load("\n".join(lines[1:end]))
    except Exception:  # noqa: BLE001 — a malformed skill must not fail the build
        return ""
    desc = (front or {}).get("description") if isinstance(front, dict) else None
    if not isinstance(desc, str):
        return ""
    # A block scalar folds into one line but keeps its newlines when literal;
    # collapse whitespace so the value fits a Markdown table cell.
    desc = " ".join(desc.split())
    # Truncate at the first sentence boundary for readability.
    idx = desc.find(". ")
    if idx != -1:
        desc = desc[: idx + 1]
    return desc


def _skill_section(name: str, org_name: str, skill: str, build: Path) -> str:
    if skill == "appsec-analyzer":
        return f"""
## `/{name}:appsec-analyzer`

Security assessment derived from the repository's code and configuration. Produces
`docs/security/threat-model.md` in your repo, checked against {org_name} security requirements.

**Depth:**

| Flag | Description |
|---|---|
| _(none)_ | Standard — full STRIDE analysis |
| `--quick` | Faster, lighter analysis |
| `--thorough` | Deepest analysis; adds architect review |

**Common options:**

| Flag | Description |
|---|---|
| `--requirements` | Check findings against {org_name} security requirements |
| `--no-requirements` | Skip requirements check for this run |
| `--resume` | Continue an interrupted assessment |
| `--rebuild` | Start completely fresh |
| `--repo <path>` | Analyze a different repository instead of the current one |
| `--output <path>` | Write results to a custom output directory |
| `--sarif` | Also write `threat-model.sarif.json` for CI/tooling integration |
| `--stop` | Stop a running assessment |

**Examples:**

```text
/{name}:appsec-analyzer
/{name}:appsec-analyzer --quick
/{name}:appsec-analyzer --thorough --requirements
/{name}:appsec-analyzer --sarif
/{name}:appsec-analyzer --repo ../other-service
```
"""
    if skill == "audit-security-requirements":
        return f"""
## `/{name}:audit-security-requirements`

Audits the entire codebase against {org_name} security requirements and verifies
whether each tagged requirement (e.g. `[SEC-AUTH-001]`) is implemented.
Prints color-coded status with evidence. Optionally saves results as JSON or Markdown.

```text
/{name}:audit-security-requirements
/{name}:audit-security-requirements --output json
```
"""
    if skill == "verify-requirements":
        return f"""
## `/{name}:verify-requirements`

Checks your recent code changes (current diff) against {org_name} security requirements.
Lighter than a full audit — scoped to what you just changed.
Use `--gate` to turn it into a CI/merge gate that fails on violations.

```text
/{name}:verify-requirements
/{name}:verify-requirements --gate
```
"""
    return ""


def _build_readme(build: Path, name: str, surface_manifest: dict, upstream_url: str | None) -> str:
    org_profile_path = build / "org-profile" / "org-profile.yaml"
    org_name = name  # fallback
    if org_profile_path.exists():
        for line in org_profile_path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if line.startswith("name:"):
                org_name = line[len("name:") :].strip()
                break

    skills = surface_manifest.get("skills", {}).get("included", [])
    hooks = surface_manifest.get("hooks", {}).get("included", [])

    # Main skill sections
    main_sections = ""
    for skill in MAIN_SKILLS:
        if skill in skills:
            main_sections += _skill_section(name, org_name, skill, build)

    # Utility skills table
    utility_rows = ""
    for skill in UTILITY_SKILLS:
        if skill in skills:
            desc = _skill_description(build, skill)
            utility_rows += f"| `/{name}:{skill}` | {desc} |\n"
    # Any skills not in either category
    known = set(MAIN_SKILLS + UTILITY_SKILLS) | INTERNAL_SKILLS
    for skill in sorted(skills):
        if skill not in known:
            desc = _skill_description(build, skill)
            utility_rows += f"| `/{name}:{skill}` | {desc} |\n"

    utility_section = ""
    if utility_rows:
        utility_section = f"""
## Utility Commands

| Command | Description |
|---|---|
{utility_rows}"""

    hooks_section = ""
    if hooks:
        hook_rows = ""
        for hook in sorted(hooks):
            desc = HOOK_DESCRIPTIONS.get(hook, "")
            hook_rows += f"| `{hook}` | {desc} |\n"
        hooks_section = f"""
## Active Hooks

| Hook | Description |
|---|---|
{hook_rows}"""

    upstream_line = ""
    if upstream_url:
        display_url = upstream_url.removesuffix(".git")
        upstream_line = f"\n- [appsec-advisor]({display_url})"

    based_on_line = ""
    if upstream_url:
        display_url = upstream_url.removesuffix(".git")
        based_on_line = f"\nBased on [appsec-advisor]({display_url})."

    readme = f"""# {name} — {org_name} AppSec Plugin for Claude Code

Internal Claude Code security plugin for {org_name}.
Runs automated threat models and security audits directly in your IDE,
with {org_name} security standards and requirements already baked in.{based_on_line}

## Getting Started

Load the plugin in any repo:

```bash
claude --plugin-dir /path/to/build/{name}
```
{main_sections}{utility_section}{hooks_section}
## Reference
{upstream_line}
"""
    return readme


def write_readme(
    build: Path, name: str, surface_manifest: dict, upstream_url: str | None, readme_path: Path | None = None
) -> None:
    content = _build_readme(build, name, surface_manifest, upstream_url)
    target = readme_path if readme_path else (build / "README.md")
    target.write_text(content, encoding="utf-8")


SOURCE_LOCK = "organization-source/components.lock.json"
SOURCE_MANIFEST = "organization-source/component-manifest.json"


def _staging_tree(root: Path) -> dict[str, bytes]:
    """Read a bounded, portable tree without following links or executing inputs."""
    import compose_plugin as composition

    files: dict[str, bytes] = {}
    seen: set[str] = set()
    total = 0

    def visit(directory: Path) -> None:
        nonlocal total
        with os.scandir(directory) as entries:
            for entry in entries:
                path = Path(entry.path)
                relative = composition.portable_path(path.relative_to(root).as_posix())
                folded = relative.casefold()
                if folded in seen or len(seen) >= composition.MAX_ENTRIES:
                    raise composition.CompositionError("staging tree has colliding paths or too many entries")
                seen.add(folded)
                mode = entry.stat(follow_symlinks=False).st_mode
                if stat.S_ISDIR(mode):
                    visit(path)
                elif stat.S_ISREG(mode):
                    data = composition.read_bounded(path, composition.MAX_ARCHIVE)
                    total += len(data)
                    if total > composition.MAX_EXPANDED:
                        raise composition.CompositionError("staging tree exceeds its byte limit")
                    files[relative] = data
                else:
                    raise composition.CompositionError("staging tree forbids links and special files")

    visit(root)
    return files


def _staging_inputs(files: dict[str, bytes]) -> tuple[dict, dict, set[str]]:
    import compose_plugin as composition
    from jsonschema import Draft202012Validator
    from referencing import Registry, Resource

    if len(files["components.lock.json"]) > 2 * 1024 * 1024 or len(files["component-manifest.json"]) > 8 * 1024 * 1024:
        raise composition.CompositionError("staging metadata exceeds its byte limit")
    lock = composition.parse_lock(files["components.lock.json"])
    manifest = json.loads(files["component-manifest.json"], object_pairs_hook=composition.unique_fields)
    schema_root = Path(__file__).resolve().parents[1] / "schemas"
    lock_schema = json.loads((schema_root / "components-lock.schema.json").read_bytes())
    schema = json.loads((schema_root / "component-manifest.schema.json").read_bytes())
    registry = Registry().with_resource("components-lock.schema.json", Resource.from_contents(lock_schema))
    if not Draft202012Validator(schema, registry=registry).is_valid(manifest):
        raise composition.CompositionError("staging manifest does not match its schema")
    if manifest["components"] != lock["components"] or manifest["lock_sha256"] != composition.digest(
        composition.canonical(lock)
    ):
        raise composition.CompositionError("staging manifest does not match its lock")
    host = next(c["id"] for c in lock["components"] if c["role"] == "host")
    owners = {}
    for component in lock["components"]:
        for mapping in component["files"]:
            owners[composition.portable_path(mapping["target"])] = component["id"]
        if component["artifact"].endswith(".whl"):
            path = "runtime/wheels/" + component["artifact"]
            owners[path] = component["id"]
            if composition.digest(files[path]) != component["sha256"]:
                raise composition.CompositionError("locked staging wheel integrity mismatch")
            members = composition.archive_files(files[path])
            for mapping in component["files"]:
                target = mapping["target"]
                if any(target == f"skills/{skill}/SKILL.md" for skill in lock.get("skill_bindings", [])):
                    target = target.removesuffix("SKILL.md") + "WORKFLOW.md"
                if files[target] != members[mapping["source"]]:
                    raise composition.CompositionError("selected engine file differs from its pinned wheel")
    for skill in lock.get("skill_bindings", []):
        prefix = f"skills/{skill}/"
        owners[prefix + "WORKFLOW.md"] = owners.pop(prefix + "SKILL.md")
        owners[prefix + "SKILL.md"] = host
    delivered = {entry["path"]: entry for entry in manifest["files"]}
    if len(delivered) != len(manifest["files"]) or set(delivered) != set(owners):
        raise composition.CompositionError("staging inventory differs from selected files")
    if set(files) != set(delivered) | composition.RESERVED:
        raise composition.CompositionError("staging tree contains missing or unlisted files")
    protected = set(composition.RESERVED)
    for path, entry in delivered.items():
        data = files[path]
        if (
            entry["component"] != owners[path]
            or entry["size"] != len(data)
            or entry["sha256"] != composition.digest(data)
        ):
            raise composition.CompositionError("staging payload integrity or ownership mismatch")
        if owners[path] != host:
            protected.add(path)
    for path in (".claude-plugin/plugin.json", "scripts/composition_runtime.py"):
        if owners.get(path) != host:
            raise composition.CompositionError("staging manifest and bootstrap must be host-owned")
    if owners.get("scripts/runtime/assessment_console.py", host) != host:
        raise composition.CompositionError("staging progress renderer must be host-owned")
    plugin = json.loads(files[".claude-plugin/plugin.json"], object_pairs_hook=composition.unique_fields)
    if not isinstance(plugin, dict) or any(plugin.get(key) != value for key, value in lock["plugin"].items()):
        raise composition.CompositionError("staging plugin identity differs from its lock")
    protected.add("scripts/composition_runtime.py")
    return lock, manifest, protected


ORG_DISPATCH = "org-profile/composition-dispatch.json"


def _staging_profile(profile: dict) -> dict:
    from composition_runtime import RuntimeError, split_profile

    try:
        return split_profile(profile)
    except RuntimeError as exc:
        _die(str(exc))


def _staging_config(files: dict[str, bytes], profile: dict) -> tuple[dict, str | None]:
    """Retain host settings and reject global settings without a migrated owner."""
    import compose_plugin as composition
    from validators.validate_config import _validate_main_config

    raw = files.get("config.json", b"{}")
    if len(raw) > 65536:
        _die("staging host configuration exceeds its byte limit")
    config = json.loads(raw, object_pairs_hook=composition.unique_fields)
    json.dumps(config, allow_nan=False)
    if not isinstance(config, dict) or set(config) - {
        "_comment",
        "banner",
        "pricing",
        "logging",
        "organization_profile",
    }:
        _die("staging global configuration has unsupported consumers; use --source")
    # The legacy validator requires external_context. Its disabled validation
    # value is never delivered; that engine-owned block is refused above.
    if _validate_main_config({**config, "external_context": {"enabled": False, "rest_url": None}}, "config.json"):
        _die("staging host configuration is invalid")
    for key, allowed in (
        ("pricing", {"input_per_1m", "output_per_1m", "cache_write_per_1m", "cache_read_per_1m"}),
        ("logging", {"max_log_bytes", "verbose"}),
    ):
        if set(config.get(key) or {}) - allowed:
            _die("staging host configuration contains unmapped nested settings")
    preset = (config.get("organization_profile") or {}).get("default_preset")
    if preset is not None and preset not in profile["presets"]:
        _die("staging configured default preset is unavailable")
    return config, preset


def _staging_profile_data(data: bytes) -> dict:
    from composition_runtime import RuntimeError, profile_data

    try:
        return profile_data(data)
    except RuntimeError as exc:
        _die(str(exc))


def _staging_readme(build: Path, name: str, skills: list[str]) -> None:
    rows = "".join(f"| `/{name}:{skill}` | {_skill_description(build, skill)} |\n" for skill in skills)
    (build / "README.md").write_text(
        f"# {name} staging plugin\n\n"
        "Unqualified organization package. This selected surface is not a replacement release of the full Advisor plugin.\n\n"
        "## Selected commands\n\n| Command | Description |\n|---|---|\n" + rows + "\n"
        "## Explicit runtime setup\n\n"
        "Use the interpreter selected by `components.lock.json`. Replace `/installed/plugin` with the extracted package directory.\n\n"
        '```sh\npython3 "/installed/plugin/scripts/composition_runtime.py" setup --plugin "/installed/plugin"\n'
        'python3 "/installed/plugin/scripts/composition_runtime.py" preflight --plugin "/installed/plugin"\n```\n\n'
        "Setup is explicit and offline. Engine invocation uses the pinned runtime; it does not resolve an engine from PATH.\n\n"
        "Engine organization settings use the public configuration API. Score has no assessment depth; unsupported settings fail packaging.\n",
        encoding="utf-8",
    )


def package_staging(args: argparse.Namespace, source: Path, org_profile: Path, build: Path, dist_dir: Path) -> None:
    """Brand the host surface, preserving pinned payloads and original provenance."""
    import compose_plugin as composition
    from validators import validate_org_profile

    composition.portable_path(args.name)
    composition.portable_path(f"{args.name}-{args.version}.tgz")
    for input_root in (source, org_profile):
        for output in (build, dist_dir):
            if output == input_root or output in input_root.parents or input_root in output.parents:
                _die("staging outputs must be separate from their inputs")
    if dist_dir == build or build in dist_dir.parents:
        _die("staging archives must be outside the build tree")
    if args.readme:
        _die("--readme is not supported with --staging; the delivered README must stay inventoried")
    files = _staging_tree(source)
    lock, manifest, protected = _staging_inputs(files)
    profile_files = _staging_tree(org_profile)
    if "org-profile.yaml" not in profile_files:
        _die("organization profile must contain org-profile.yaml")
    policy, policy_path = load_package_policy(org_profile, args.package_policy)
    profile = _staging_profile_data(profile_files["org-profile.yaml"])
    # Shape validation remains mandatory even when optional semantic checks are skipped.
    errors = validate_org_profile._schema_errors(profile, validate_org_profile._load_schema())
    if errors:
        _die("staging organization profile does not match its schema")
    if (
        ORG_DISPATCH in files
        or "org-profile/analyzer-config.yaml" in files
        or {"composition-dispatch.json", "analyzer-config.yaml"} & profile_files.keys()
    ):
        _die("organization input collides with generated dispatch")
    if validate_org_profile._check_default_preset(profile) or validate_org_profile._check_compatibility(
        profile, lock["plugin"]["version"]
    ):
        _die("organization preset or Advisor compatibility is invalid")
    _, configured_preset = _staging_config(files, profile)
    dispatch = _staging_profile(profile)
    if configured_preset is not None:
        dispatch["analyzer"]["default_preset"] = configured_preset
    engine = next(component for component in lock["components"] if component["role"] == "engine")
    if "organization_config" not in engine["capabilities"]:
        _die("organization dispatch requires a new Analyzer pin declaring organization_config; reassemble first")
    engine_config = {
        "schema_version": 1,
        "engine_version": engine["version"],
        "advisor_core_version": lock["plugin"]["version"],
        "profile": {**dispatch["identity"], **dispatch["analyzer"]},
    }
    import jsonschema

    engine_members = composition.archive_files(files["runtime/wheels/" + engine["artifact"]])
    schema_bytes = engine_members.get("appsec_analyzer/schemas/organization-config.schema.json")
    if schema_bytes is None:
        _die("pinned Analyzer is missing its public organization configuration schema")
    if len(schema_bytes) > 65536:
        _die("pinned organization schema exceeds its byte limit")
    engine_schema = json.loads(schema_bytes, object_pairs_hook=composition.unique_fields)
    # Public schema v1 is self-contained. No schema-controlled fetch or resolver
    # is allowed while packaging an otherwise offline composition.
    pending = [engine_schema]
    while pending:
        node = pending.pop()
        if isinstance(node, dict):
            if {"$ref", "$dynamicRef"} & node.keys():
                _die("pinned organization schema must not select external resources")
            pending.extend(node.values())
        elif isinstance(node, list):
            pending.extend(node)
    if len(composition.canonical(engine_config)) > 65536 or not jsonschema.Draft202012Validator(engine_schema).is_valid(
        engine_config
    ):
        _die("engine organization projection violates the pinned public schema")
    # Packaging never updates a delivered bootstrap. New dispatch requires a
    # newly assembled host artifact containing the reviewed consumer.
    bootstrap = Path(__file__).with_name("composition_runtime.py").read_bytes()
    if files.get("scripts/composition_runtime.py") != bootstrap:
        _die(
            "organization dispatch requires reassembly with the current pinned Advisor host; bootstrap was not changed"
        )
    dispatch_schema = json.loads((Path(__file__).parents[1] / "schemas/organization-dispatch.schema.json").read_bytes())
    if not jsonschema.Draft202012Validator(dispatch_schema).is_valid(dispatch):
        _die("organization dispatch does not match its schema")
    upstream_url = args.upstream_url or next(c["source"] for c in lock["components"] if c["role"] == "host")
    build.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix=".organization-package-", dir=build.parent) as temporary:
        work = Path(temporary) / "plugin"
        work.mkdir()
        for path, data in files.items():
            target = work / path
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
        overlay = work / "org-profile"
        if overlay.exists():
            shutil.rmtree(overlay)
        for path, data in profile_files.items():
            target = overlay / path
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
        if validate_org_profile._check_hooks(profile, overlay) or validate_org_profile._check_mcp(profile):
            _die("staging organization hooks or MCP declarations violate their security contract")
        host_component = next(c for c in lock["components"] if c["role"] == "host")
        patch_plugin_json(
            work,
            args.name,
            args.version,
            args.description,
            source,
            provenance={
                "appsec_advisor_core_version": host_component["version"],
                "appsec_advisor_core_commit": host_component["revision"],
            },
        )
        if not (work / "config.json").exists():
            (work / "config.json").write_text("{}\n", encoding="utf-8")
        patch_config(work, args.info_url)
        if configured_preset is not None:
            host_config = json.loads((work / "config.json").read_bytes())
            host_config["organization_profile"]["default_preset"] = configured_preset
            (work / "config.json").write_bytes(composition.canonical(host_config))
        (work / "hooks").mkdir(exist_ok=True)
        apply_package_surface_policy(work, policy, policy_path, upstream_url, required_skills=set())
        surface = json.loads((work / SURFACE_MANIFEST).read_bytes())
        # Until the host policy gate migrates, physically exclude disabled entries.
        known = set(surface["skills"]["included"]) | set(surface["skills"]["removed"])
        for skill, toggle in _org_profile_skill_toggles(work).items():
            if skill not in known:
                _die("staging skill toggle names an unavailable skill")
            if not toggle["enabled"] and skill in surface["skills"]["included"]:
                if skill in _required_internal_skills(work):
                    _die("staging skill toggle disables an agent's required internal skill")
                shutil.rmtree(work / "skills" / skill)
                surface["skills"]["included"].remove(skill)
                surface["skills"]["removed"].append(skill)
        surface["skills"]["removed"].sort()
        surface["composition"] = {"source_lock": SOURCE_LOCK, "source_manifest": SOURCE_MANIFEST}
        (work / SURFACE_MANIFEST).write_bytes(composition.canonical(surface))
        if not args.skip_validation:
            errors = validate_org_profile.validate(profile, overlay, lock["plugin"]["version"], plugin_root=work)
            if errors:
                _die("staging organization profile semantic validation failed")
        (work / ORG_DISPATCH).write_bytes(composition.canonical(dispatch))
        (work / "org-profile/analyzer-config.yaml").write_bytes(composition.canonical(engine_config))
        _staging_readme(work, args.name, surface["skills"]["included"])
        if args.name != UPSTREAM_NAMESPACE:
            dispatch_protected = protected | {ORG_DISPATCH, "org-profile/analyzer-config.yaml"}
            rewrite_namespace(work, args.name, protected=dispatch_protected)
            check_namespace_leaks(work, protected=dispatch_protected)
        result = _staging_tree(work)
        removed = set(files) - set(result)
        for path in protected:
            # Surface selection may remove an entire skill, never individual resources.
            parts = path.split("/")
            excluded_skill = len(parts) > 2 and parts[0] == "skills" and parts[1] in surface["skills"]["removed"]
            if path in removed and excluded_skill:
                continue
            if result.get(path) != files[path]:
                _die("organization packaging attempted to change a pinned component file")
        for path in (SOURCE_LOCK, SOURCE_MANIFEST):
            if path in result:
                _die("staging input collides with organization provenance output")
        result[SOURCE_LOCK] = files["components.lock.json"]
        result[SOURCE_MANIFEST] = files["component-manifest.json"]
        included = set(surface["skills"]["included"])
        if "skill_bindings" in lock:
            lock["skill_bindings"] = [skill for skill in lock["skill_bindings"] if skill in included]
        result["components.lock.json"] = composition.canonical(lock)
        host = next(c["id"] for c in lock["components"] if c["role"] == "host")
        original = {entry["path"]: entry for entry in manifest["files"]}
        manifest["lock_sha256"] = composition.digest(result["components.lock.json"])
        manifest["files"] = [
            {
                "path": path,
                "sha256": composition.digest(data),
                "size": len(data),
                "component": original.get(path, {}).get("component", host),
            }
            for path, data in sorted(result.items())
            if path not in composition.RESERVED
        ]
        result["component-manifest.json"] = composition.canonical(manifest)
        if len(result["component-manifest.json"]) > 8 * 1024 * 1024:
            _die("organization package manifest exceeds the bootstrap's byte limit")
        for path, data in result.items():
            target = work / path
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
        # Recheck all output bounds, including generated metadata, before publication.
        _staging_tree(work)
        remove_stale_archive(args.name, args.version, dist_dir)
        if build.exists():
            shutil.rmtree(build)
        work.replace(build)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build and validate an internal appsec-advisor plugin package.")
    inputs = parser.add_mutually_exclusive_group()
    inputs.add_argument("--source", default=".", help="upstream appsec-advisor checkout")
    inputs.add_argument("--staging", help="extracted, verified compose_plugin staging tree")
    parser.add_argument("--org-profile", required=True, help="org-profile directory to bundle")
    parser.add_argument("--name", required=True, help="internal plugin name / command namespace")
    parser.add_argument("--version", required=True, help="internal package version")
    parser.add_argument("--build-dir", default="build", help="build output directory")
    parser.add_argument("--dist-dir", default="dist", help="tarball output directory")
    parser.add_argument("--description", default=None, help="plugin.json description override")
    parser.add_argument(
        "--upstream-url", default=None, help="upstream plugin repository URL recorded in package-surface.json"
    )
    parser.add_argument(
        "--info-url",
        default=None,
        help="internal documentation URL shown in the session banner; pass an empty value to drop the line",
    )
    parser.add_argument(
        "--readme", default=None, help="write generated README.md to this path (default: inside build tree)"
    )
    parser.add_argument(
        "--skip-validation",
        action="store_true",
        help="skip packaged config and org-profile validation",
    )
    parser.add_argument(
        "--skip-archive",
        action="store_true",
        help="build the packaged tree but do not create a tarball",
    )
    parser.add_argument(
        "--package-policy",
        default=None,
        help=("optional package surface policy; defaults to org-profile/package-policy.yaml when present"),
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    source = Path(args.source).resolve()
    org_profile = Path(args.org_profile).resolve()
    build = (Path(args.build_dir) / args.name).resolve()
    dist_dir = Path(args.dist_dir).resolve()
    _validate_package_name(args.name)
    _validate_version(args.version)
    if args.staging:
        import compose_plugin as composition

        try:
            package_staging(args, Path(args.staging).resolve(), org_profile, build, dist_dir)
        except (composition.CompositionError, OSError, ValueError, KeyError, TypeError, RecursionError):
            _die("invalid or incompatible staging package; no artifact published")
        if not args.skip_archive:
            tar_path, sha_path = write_archive(build, args.name, args.version, dist_dir, deterministic=True)
            print(f"==> Artifact: {tar_path}\n==> SHA256:   {sha_path}")
        print(f"==> Build tree ready at {build}")
        return 0
    package_policy, package_policy_path = load_package_policy(org_profile, args.package_policy)

    _validate_package_name(args.name)
    _validate_version(args.version)
    _require_plugin_root(source)
    remove_stale_archive(args.name, args.version, dist_dir)

    print(f"==> Packaging {args.name} {args.version}", flush=True)
    upstream_url = args.upstream_url or upstream_url_from_git(source)
    copy_source(source, build)
    overlay_org_profile(org_profile, build)
    patch_plugin_json(build, args.name, args.version, args.description, source)
    patch_config(build, args.info_url)
    apply_package_surface_policy(build, package_policy, package_policy_path, upstream_url)
    surface_manifest = json.loads((build / SURFACE_MANIFEST).read_text(encoding="utf-8"))
    readme_path = Path(args.readme) if args.readme else None
    write_readme(build, args.name, surface_manifest, upstream_url, readme_path)
    rewrite_namespace(build, args.name)
    check_namespace_leaks(build)

    if not args.skip_validation:
        run_validation(build)

    if args.skip_archive:
        print(f"==> Build tree ready at {build}")
    else:
        tar_path, sha_path = write_archive(build, args.name, args.version, dist_dir)
        print(f"==> Build tree ready at {build}")
        print(f"==> Artifact: {tar_path}")
        print(f"==> SHA256:   {sha_path}")
    return 0


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