"""Dependency risk dashboard -- pin coverage, risk scoring, supply-chain health."""

from __future__ import annotations

import re
from collections import Counter
from pathlib import Path
from typing import NamedTuple

import click

# Pre-compiled regex for requirements.txt parsing
_RE_INLINE_COMMENT = re.compile(r"\s+#.*$")
_RE_REQUIREMENT = re.compile(r"^([A-Za-z0-9_.\-]+)(\[.*?\])?(.*)?$")

from roam import __version__
from roam.capability import roam_capability
from roam.db.connection import find_project_root
from roam.output.formatter import format_table, json_envelope, to_json
from roam.output.sarif import to_sarif, write_sarif


class Dependency(NamedTuple):
    name: str
    version_spec: str
    is_dev: bool
    pin_status: str
    risk_level: str
    source_file: str
    ecosystem: str


def _pin_status_python(spec: str) -> str:
    if not spec or spec.strip() in ("*", ""):
        return "unpinned"
    if spec.startswith("=="):
        return "exact"
    if spec.startswith("~="):
        return "range"
    if "*" in spec:
        return "unpinned"
    if re.search(r"[><!]", spec):
        return "range"
    return "unpinned"


def _pin_status_npm(spec: str) -> str:
    if not spec or spec in ("*", "latest", "x", ""):
        return "unpinned"
    if re.match(r"^=?[0-9]", spec):
        return "exact"
    if spec.startswith(("^", "~")):
        return "range"
    if spec.startswith((">", "<")):
        return "range"
    if re.search(r"[*xX]", spec):
        return "unpinned"
    return "range"


def _pin_status_go(spec: str) -> str:
    if not spec:
        return "unpinned"
    if re.match(r"v\d+\.\d+\.\d+-\d{14}-[0-9a-f]+", spec):
        return "exact"
    if re.match(r"v\d+\.\d+\.\d+", spec):
        return "exact"
    return "range"


def _pin_status_rust(spec: str) -> str:
    if not spec or spec in ("*", ""):
        return "unpinned"
    if spec.startswith("=") and not spec.startswith(">="):
        return "exact"
    if spec.startswith(("^", "~")):
        return "range"
    if re.match(r"^\d", spec):
        return "range"
    if re.search(r"[><!]", spec):
        return "range"
    if "*" in spec:
        return "unpinned"
    return "range"


def _pin_status_ruby(spec: str) -> str:
    if not spec:
        return "unpinned"
    if re.match(r"^=\s*\d", spec):
        return "exact"
    if spec.startswith("~>"):
        return "range"
    if re.search(r"[><!]", spec):
        return "range"
    return "unpinned"


def _risk_level(pin_status: str, is_dev: bool) -> str:
    level = {"exact": "low", "range": "medium", "unpinned": "high"}[pin_status]
    # Dev dependencies don't ship to production — reduce risk by one level
    if is_dev:
        level = {"high": "medium", "medium": "low", "low": "low"}[level]
    return level


def _parse_requirements_txt(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or line.startswith("-"):
            continue
        line = _RE_INLINE_COMMENT.sub("", line)
        m = _RE_REQUIREMENT.match(line)
        if not m:
            continue
        name = m.group(1).strip()
        spec = (m.group(3) or "").strip()
        pin = _pin_status_python(spec)
        deps.append(
            Dependency(
                name=name,
                version_spec=spec,
                is_dev=False,
                pin_status=pin,
                risk_level=_risk_level(pin, False),
                source_file=source,
                ecosystem="python",
            )
        )
    return deps


_SETUP_INSTALL_REQUIRES_RE = re.compile(r"install_requires\s*=\s*\[([^\]]*)\]", re.DOTALL)


def _parse_setup_py(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    m = _SETUP_INSTALL_REQUIRES_RE.search(text)
    if m:
        for raw in re.findall(r"[\x27\x22]([^\x27\x22]+)[\x27\x22]", m.group(1)):
            name_m = re.match(r"^([A-Za-z0-9_.\-]+)(.*)?$", raw.strip())
            if not name_m:
                continue
            name = name_m.group(1)
            spec = name_m.group(2).strip() if name_m.group(2) else ""
            pin = _pin_status_python(spec)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=spec,
                    is_dev=False,
                    pin_status=pin,
                    risk_level=_risk_level(pin, False),
                    source_file=source,
                    ecosystem="python",
                )
            )
    return deps


def _parse_setup_cfg(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    in_install = False
    in_continuation = False
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("["):
            in_install = stripped == "[options]"
            in_continuation = False
            continue
        if in_install:
            if re.match(r"install_requires\s*=", stripped):
                in_continuation = True
                continue
            if in_continuation and line.startswith((" ", "	")) and stripped and not stripped.startswith(";"):
                m2 = re.match(r"^([A-Za-z0-9_.\-]+)(.*)?$", stripped)
                if m2:
                    name = m2.group(1)
                    spec = m2.group(2).strip() if m2.group(2) else ""
                    pin = _pin_status_python(spec)
                    deps.append(
                        Dependency(
                            name=name,
                            version_spec=spec,
                            is_dev=False,
                            pin_status=pin,
                            risk_level=_risk_level(pin, False),
                            source_file=source,
                            ecosystem="python",
                        )
                    )
            elif stripped and not line.startswith((" ", "	")):
                in_continuation = False
    return deps


def _parse_pyproject_toml(path: Path, source: str) -> list[Dependency]:
    """Parse pyproject.toml -- [project] PEP 621 and [tool.poetry] formats."""
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps

    # Strip TOML comments line-by-line so apostrophes inside English
    # comments (e.g. ``# when these aren't installed``) don't open
    # fake quoted strings for the regex below to match. A line-by-line
    # strip is safe because TOML comments end at end-of-line.
    text = "\n".join(re.sub(r"\s#.*$|^\s*#.*$", "", line) for line in text.splitlines())

    # PEP 621 [project] format
    sec_m = re.search(r"^\[project\]", text, re.MULTILINE)
    if sec_m:
        dep_m = re.search(
            r"^dependencies\s*=\s*\[([^\]]*)\]",
            text[sec_m.start() :],
            re.MULTILINE | re.DOTALL,
        )
        if dep_m:
            for raw in re.findall(r"[\x27\x22]([^\x27\x22]+)[\x27\x22]", dep_m.group(1)):
                raw = raw.strip()
                nm = re.match(r"^([A-Za-z0-9_.\-]+)(\[.*?\])?(.*)?$", raw)
                if nm:
                    name = nm.group(1)
                    spec = (nm.group(3) or "").strip()
                    pin = _pin_status_python(spec)
                    deps.append(
                        Dependency(
                            name=name,
                            version_spec=spec,
                            is_dev=False,
                            pin_status=pin,
                            risk_level=_risk_level(pin, False),
                            source_file=source,
                            ecosystem="python",
                        )
                    )
        opt_m = re.search(
            r"^\[project\.optional-dependencies\](.*?)(?=^\[|\Z)",
            text,
            re.MULTILINE | re.DOTALL,
        )
        if opt_m:
            for raw in re.findall(r"[\x27\x22]([^\x27\x22]+)[\x27\x22]", opt_m.group(1)):
                raw = raw.strip()
                nm = re.match(r"^([A-Za-z0-9_.\-]+)(\[.*?\])?(.*)?$", raw)
                if nm:
                    name = nm.group(1)
                    spec = (nm.group(3) or "").strip()
                    pin = _pin_status_python(spec)
                    deps.append(
                        Dependency(
                            name=name,
                            version_spec=spec,
                            is_dev=True,
                            pin_status=pin,
                            risk_level=_risk_level(pin, True),
                            source_file=source,
                            ecosystem="python",
                        )
                    )

    # [tool.poetry.dependencies]
    poetry_m = re.search(r"^\[tool\.poetry\.dependencies\](.*?)(?=^\[|\Z)", text, re.MULTILINE | re.DOTALL)
    if poetry_m:
        for line in poetry_m.group(1).splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            kv = re.match(r"^([A-Za-z0-9_.\-]+)\s*=\s*(.+)$", line)
            if not kv:
                continue
            name = kv.group(1)
            if name.lower() == "python":
                continue
            val = kv.group(2).strip().strip('"').strip("'")
            tv = re.search(r"version\s*=\s*[\x27\x22]([^\x27\x22]+)[\x27\x22]", val)
            if tv:
                val = tv.group(1)
            pin = _pin_status_python(val)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=val,
                    is_dev=False,
                    pin_status=pin,
                    risk_level=_risk_level(pin, False),
                    source_file=source,
                    ecosystem="python",
                )
            )

    # [tool.poetry.dev-dependencies]
    poetry_dev_m = re.search(r"^\[tool\.poetry\.dev-dependencies\](.*?)(?=^\[|\Z)", text, re.MULTILINE | re.DOTALL)
    if poetry_dev_m:
        for line in poetry_dev_m.group(1).splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            kv = re.match(r"^([A-Za-z0-9_.\-]+)\s*=\s*(.+)$", line)
            if not kv:
                continue
            name = kv.group(1)
            val = kv.group(2).strip().strip('"').strip("'")
            tv = re.search(r"version\s*=\s*[\x27\x22]([^\x27\x22]+)[\x27\x22]", val)
            if tv:
                val = tv.group(1)
            pin = _pin_status_python(val)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=val,
                    is_dev=True,
                    pin_status=pin,
                    risk_level=_risk_level(pin, True),
                    source_file=source,
                    ecosystem="python",
                )
            )

    return deps


def _parse_package_json(path: Path, source: str) -> list[Dependency]:
    import json as _json

    deps: list[Dependency] = []
    try:
        data = _json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except (OSError, ValueError):
        return deps
    for section, is_dev in [
        ("dependencies", False),
        ("devDependencies", True),
        ("peerDependencies", True),
        ("optionalDependencies", True),
    ]:
        for name, spec in (data.get(section) or {}).items():
            if not isinstance(spec, str):
                continue
            pin = _pin_status_npm(spec)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=spec,
                    is_dev=is_dev,
                    pin_status=pin,
                    risk_level=_risk_level(pin, is_dev),
                    source_file=source,
                    ecosystem="javascript",
                )
            )
    return deps


def _parse_go_mod(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    for block in re.findall(r"require\s*\((.*?)\)", text, re.DOTALL):
        for line in block.splitlines():
            line = line.strip()
            if not line or line.startswith("//"):
                continue
            parts = line.split()
            if len(parts) >= 2:
                name, ver = parts[0], parts[1]
                is_indirect = "// indirect" in line
                pin = _pin_status_go(ver)
                deps.append(
                    Dependency(
                        name=name,
                        version_spec=ver,
                        is_dev=is_indirect,
                        pin_status=pin,
                        risk_level=_risk_level(pin, is_indirect),
                        source_file=source,
                        ecosystem="go",
                    )
                )
    for m in re.finditer(r"^require\s+(\S+)\s+(\S+)", text, re.MULTILINE):
        name, ver = m.group(1), m.group(2)
        pin = _pin_status_go(ver)
        deps.append(
            Dependency(
                name=name,
                version_spec=ver,
                is_dev=False,
                pin_status=pin,
                risk_level=_risk_level(pin, False),
                source_file=source,
                ecosystem="go",
            )
        )
    return deps


def _parse_cargo_toml(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    section_re = re.compile(r"^\[([^\]]+)\]", re.MULTILINE)
    sections = [(m.group(1), m.start()) for m in section_re.finditer(text)]

    def _get_section_body(section_name: str) -> str:
        for i, (sname, start) in enumerate(sections):
            if sname.strip() == section_name:
                end = sections[i + 1][1] if i + 1 < len(sections) else len(text)
                return text[start:end]
        return ""

    for sec_name, is_dev in [
        ("dependencies", False),
        ("dev-dependencies", True),
        ("build-dependencies", True),
    ]:
        body = _get_section_body(sec_name)
        if not body:
            continue
        for line in body.splitlines():
            line = line.strip()
            if not line or line.startswith("#") or line.startswith("["):
                continue
            kv = re.match(r"^([A-Za-z0-9_\-]+)\s*=\s*(.+)$", line)
            if not kv:
                continue
            name = kv.group(1)
            val = kv.group(2).strip()
            tv = re.search(r"version\s*=\s*[\x27\x22]([^\x27\x22]*)[\x27\x22]", val)
            if tv:
                spec = tv.group(1)
            else:
                spec = val.strip(chr(34)).strip(chr(39))
            pin = _pin_status_rust(spec)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=spec,
                    is_dev=is_dev,
                    pin_status=pin,
                    risk_level=_risk_level(pin, is_dev),
                    source_file=source,
                    ecosystem="rust",
                )
            )
    return deps


def _parse_pom_xml(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    for dep_block in re.findall(r"<dependency>(.*?)</dependency>", text, re.DOTALL):
        group_m = re.search(r"<groupId>([^<]+)</groupId>", dep_block)
        art_m = re.search(r"<artifactId>([^<]+)</artifactId>", dep_block)
        ver_m = re.search(r"<version>([^<]+)</version>", dep_block)
        scope_m = re.search(r"<scope>([^<]+)</scope>", dep_block)
        if not (group_m and art_m):
            continue
        name = f"{group_m.group(1).strip()}:{art_m.group(1).strip()}"
        spec = ver_m.group(1).strip() if ver_m else ""
        scope = (scope_m.group(1).strip() if scope_m else "compile").lower()
        is_dev = scope in ("test", "provided", "system")
        pin = "exact" if spec and re.match(r"^\d+\.\d+", spec) else _pin_status_python(spec)
        deps.append(
            Dependency(
                name=name,
                version_spec=spec,
                is_dev=is_dev,
                pin_status=pin,
                risk_level=_risk_level(pin, is_dev),
                source_file=source,
                ecosystem="java",
            )
        )
    return deps


def _parse_build_gradle(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    for m in re.finditer(r"(\w+)\s+[\x27\x22]([^\x27\x22]+:[^\x27\x22]+:[^\x27\x22]+)[\x27\x22]", text):
        config = m.group(1).lower()
        parts = m.group(2).split(":")
        if len(parts) < 3:
            continue
        name = f"{parts[0]}:{parts[1]}"
        spec = parts[2]
        is_dev = config in (
            "testimplementation",
            "testcompile",
            "testruntime",
            "debugimplementation",
            "androidtestimplementation",
        )
        pin = "exact" if re.match(r"^\d+\.\d+", spec) else "range"
        deps.append(
            Dependency(
                name=name,
                version_spec=spec,
                is_dev=is_dev,
                pin_status=pin,
                risk_level=_risk_level(pin, is_dev),
                source_file=source,
                ecosystem="java",
            )
        )
    return deps


def _pin_status_composer(spec: str) -> str:
    """Pin-status classifier for composer (PHP) version specifiers."""
    if not spec or spec.strip() in ("*", ""):
        return "unpinned"
    s = spec.strip()
    # Exact version: "1.2.3" or "=1.2.3" or "v1.2.3"
    if re.match(r"^=?v?\d+\.\d+\.\d+", s):
        return "exact"
    if s.startswith(("^", "~")):
        return "range"
    if re.search(r"[><!]", s):
        return "range"
    if "*" in s:
        return "unpinned"
    return "range"


def _parse_composer_json(path: Path, source: str) -> list[Dependency]:
    """Parse a composer.json file -- `require` and `require-dev` sections."""
    import json as _json

    deps: list[Dependency] = []
    try:
        data = _json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except (OSError, ValueError):
        return deps
    if not isinstance(data, dict):
        return deps
    for section, is_dev in [("require", False), ("require-dev", True)]:
        block = data.get(section, {})
        if not isinstance(block, dict):
            continue
        for name, spec in block.items():
            if not isinstance(name, str):
                continue
            # Skip the PHP runtime + ext-* platform pseudo-packages
            if name.lower() == "php" or name.lower().startswith("ext-"):
                continue
            spec_str = spec if isinstance(spec, str) else ""
            pin = _pin_status_composer(spec_str)
            deps.append(
                Dependency(
                    name=name,
                    version_spec=spec_str,
                    is_dev=is_dev,
                    pin_status=pin,
                    risk_level=_risk_level(pin, is_dev),
                    source_file=source,
                    ecosystem="php",
                )
            )
    return deps


def _parse_gemfile(path: Path, source: str) -> list[Dependency]:
    deps: list[Dependency] = []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return deps
    in_test_group = False
    for line in text.splitlines():
        stripped = line.strip()
        if re.match(r"group\s+.*:(?:test|development|staging)", stripped):
            in_test_group = True
        if stripped == "end" and in_test_group:
            in_test_group = False
            continue
        if stripped.startswith("#"):
            continue
        m = re.match(
            r"gem\s+[\x27\x22]([^\x27\x22]+)[\x27\x22](?:,\s*[\x27\x22]([^\x27\x22]*)[\x27\x22])?",
            stripped,
        )
        if not m:
            continue
        name = m.group(1)
        spec = m.group(2) or ""
        pin = _pin_status_ruby(spec)
        deps.append(
            Dependency(
                name=name,
                version_spec=spec,
                is_dev=in_test_group,
                pin_status=pin,
                risk_level=_risk_level(pin, in_test_group),
                source_file=source,
                ecosystem="ruby",
            )
        )
    return deps


_DEP_FILES: dict[str, tuple] = {
    "requirements.txt": (_parse_requirements_txt, "python"),
    "requirements-dev.txt": (_parse_requirements_txt, "python"),
    "requirements_dev.txt": (_parse_requirements_txt, "python"),
    "requirements-test.txt": (_parse_requirements_txt, "python"),
    "constraints.txt": (_parse_requirements_txt, "python"),
    "setup.py": (_parse_setup_py, "python"),
    "setup.cfg": (_parse_setup_cfg, "python"),
    "pyproject.toml": (_parse_pyproject_toml, "python"),
    "package.json": (_parse_package_json, "javascript"),
    "go.mod": (_parse_go_mod, "go"),
    "Cargo.toml": (_parse_cargo_toml, "rust"),
    "pom.xml": (_parse_pom_xml, "java"),
    "build.gradle": (_parse_build_gradle, "java"),
    "build.gradle.kts": (_parse_build_gradle, "java"),
    "Gemfile": (_parse_gemfile, "ruby"),
    "composer.json": (_parse_composer_json, "php"),
}


_DEV_FILE_MARKERS = frozenset(
    [
        "requirements-dev.txt",
        "requirements_dev.txt",
        "requirements-test.txt",
        "requirements-ci.txt",
    ]
)


_SUBDIR_DEP_FILES: frozenset[str] = frozenset({"composer.json", "package.json", "go.mod", "Cargo.toml", "Gemfile"})

_SUBDIR_SKIP: frozenset[str] = frozenset(
    {"node_modules", ".git", ".roam", "vendor", "dist", "build", "out", "coverage"}
)


def discover_and_parse(project_root: Path) -> list[Dependency]:
    """Discover dependency manifests at the project root + 1-deep subdirs.

    Root-level files are always scanned. For monorepos and backend/frontend
    splits, we also scan 1-level subdirectories for a curated set of
    manifest types (``composer.json``, ``package.json``, ``go.mod``,
    ``Cargo.toml``, ``Gemfile``) so that, e.g., ``./backend/composer.json``
    isn't invisible to ``supply-chain`` / ``sbom``.
    """
    all_deps: list[Dependency] = []

    # Root-level discovery (existing behaviour)
    for filename, (parser_fn, _eco) in _DEP_FILES.items():
        candidate = project_root / filename
        if candidate.is_file():
            parsed = parser_fn(candidate, filename)
            if filename in _DEV_FILE_MARKERS:
                parsed = [d._replace(is_dev=True) for d in parsed]
            all_deps.extend(parsed)

    # 1-deep subdir discovery for curated manifest types
    if project_root.is_dir():
        for sub in project_root.iterdir():
            if not sub.is_dir():
                continue
            if sub.name in _SUBDIR_SKIP or sub.name.startswith("."):
                continue
            for filename, (parser_fn, _eco) in _DEP_FILES.items():
                if filename not in _SUBDIR_DEP_FILES:
                    continue
                candidate = sub / filename
                if candidate.is_file():
                    rel_source = f"{sub.name}/{filename}"
                    parsed = parser_fn(candidate, rel_source)
                    all_deps.extend(parsed)

    seen: set[tuple[str, str, str]] = set()
    unique: list[Dependency] = []
    for dep in all_deps:
        key = (dep.name.lower(), dep.ecosystem, dep.source_file)
        if key not in seen:
            seen.add(key)
            unique.append(dep)
    return unique


def compute_risk_score(deps: list[Dependency]) -> dict:
    if not deps:
        return dict(
            score=100,
            pin_coverage=1.0,
            dev_ratio=0.0,
            total=0,
            direct_count=0,
            dev_count=0,
            exact_count=0,
            range_count=0,
            unpinned_count=0,
        )
    total = len(deps)
    direct = [d for d in deps if not d.is_dev]
    dev = [d for d in deps if d.is_dev]
    exact = sum(1 for d in deps if d.pin_status == "exact")
    rng = sum(1 for d in deps if d.pin_status == "range")
    unpinned = sum(1 for d in deps if d.pin_status == "unpinned")
    pin_coverage = exact / total
    dev_ratio = len(dev) / total
    diversity_penalty = 1.0 - (unpinned / total)
    raw = pin_coverage * 0.6 + dev_ratio * 0.2 + diversity_penalty * 0.2
    score = max(0, min(100, round(raw * 100)))
    return dict(
        score=score,
        pin_coverage=round(pin_coverage, 4),
        dev_ratio=round(dev_ratio, 4),
        total=total,
        direct_count=len(direct),
        dev_count=len(dev),
        exact_count=exact,
        range_count=rng,
        unpinned_count=unpinned,
    )


def top_risky(deps: list[Dependency], n: int = 5) -> list[Dependency]:
    _pin_rank = {"unpinned": 2, "range": 1, "exact": 0}
    return sorted(deps, key=lambda d: (_pin_rank[d.pin_status], not d.is_dev), reverse=True)[:n]


def supply_chain_to_sarif(deps: list[Dependency], score: int) -> dict:
    rules = [
        {
            "id": "supply-chain/unpinned-dependency",
            "shortDescription": "Dependency has no version pin",
            "helpUri": "https://github.com/AbanteAI/roam-code#supply-chain",
            "defaultLevel": "warning",
        },
        {
            "id": "supply-chain/range-dependency",
            "shortDescription": "Dependency uses a version range instead of exact pin",
            "helpUri": "https://github.com/AbanteAI/roam-code#supply-chain",
            "defaultLevel": "note",
        },
    ]
    results = []
    for dep in deps:
        if dep.pin_status == "exact":
            continue
        rule_id = (
            "supply-chain/unpinned-dependency" if dep.pin_status == "unpinned" else "supply-chain/range-dependency"
        )
        level = "warning" if dep.pin_status == "unpinned" else "note"
        spec_str = f" ({dep.version_spec})" if dep.version_spec else ""
        dev_str = " [dev]" if dep.is_dev else ""
        results.append(
            {
                "ruleId": rule_id,
                "level": level,
                "message": {
                    "text": (
                        f"{dep.name}{spec_str} in {dep.source_file} -- {dep.pin_status}{dev_str} ({dep.ecosystem})"
                    )
                },
                "locations": [{"physicalLocation": {"artifactLocation": {"uri": dep.source_file.replace("\\", "/")}}}],
            }
        )
    return to_sarif("roam-code", __version__, rules, results)


@roam_capability(
    name="supply-chain",
    category="reports",
    summary="Dependency risk dashboard: pin coverage, risk scoring, supply-chain health",
    maturity="stable",
    mcp_expose=True,
    mcp_preset=("core", "compliance"),
    side_effect=False,
    task_required=False,
    destructive=False,
    stale_sensitive=True,
    ai_safe=True,
    requires_index=True,
)
@click.command("supply-chain")
@click.option(
    "--top", "--limit", "top", default=5, show_default=True, help="Number of riskiest dependencies to highlight"
)  # W1142: --limit alias
@click.pass_context
def supply_chain(ctx, top):
    """Dependency risk dashboard: pin coverage, risk scoring, supply-chain health.

    Unlike ``sbom`` (which generates a software bill of materials), this command
    analyzes dependency manifest files across 7 ecosystems for pin coverage --
    exact pins, range constraints, and unpinned dependencies.
    """
    json_mode = ctx.obj.get("json") if ctx.obj else False
    sarif_mode = ctx.obj.get("sarif") if ctx.obj else False
    token_budget = ctx.obj.get("budget", 0) if ctx.obj else 0

    # W607-AK -- substrate-boundary plumbing for the SBOM/VEX projection
    # leg of the W805 cross-artifact-consistency family. Prior to W607-AK a
    # raise inside any of discover_and_parse / compute_risk_score / top_risky
    # / supply_chain_to_sarif / write_sarif crashed the whole supply-chain
    # build wholesale. Each is wrapped via ``_run_check_ak`` so a raise
    # becomes a structured
    # ``supply_chain_<phase>_failed:<exc_class>:<detail>`` marker on
    # ``_w607ak_warnings_out`` -- the envelope still emits cleanly with
    # whatever signal the remaining substrates produced.
    #
    # cmd_supply_chain is the SBOM/VEX projection on the W805 cross-artifact
    # family (CGA/VSA -> SBOM/VEX projection chain) -- producer-side consumer
    # of vuln-reach / taint reachability relevant to the W805 6-artifact
    # identity-coherence story.
    #
    # Marker prefix discipline: every W607-AK substrate marker uses the
    # canonical ``supply_chain_<phase>_failed:<exc_class>:<detail>`` shape.
    # cmd_supply_chain has only a thin pre-existing ``warnings_out`` channel
    # (W1142-followup-B cap-hit disclosure) -- W607-AK is ADDITIVE: the
    # accumulator-based markers merge into the SAME ``summary.warnings_out``
    # field. The marker prefix disambiguates them downstream
    # (``truncated to ...`` vs ``supply_chain_<phase>_failed:*``).
    _w607ak_warnings_out: list[str] = []

    def _run_check_ak(phase: str, fn, *args, default=None, **kwargs):
        """Run one substrate helper with W607-AK marker emission.

        On a clean call the result is returned as-is. On an uncaught
        exception, surface a ``supply_chain_<phase>_failed:<exc_class>:<detail>``
        marker via ``_w607ak_warnings_out`` and return *default* -- the
        envelope still emits cleanly with the remaining substrates.
        """
        try:
            return fn(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 -- top-level disclosure
            _w607ak_warnings_out.append(f"supply_chain_{phase}_failed:{type(exc).__name__}:{exc}")
            return default

    # W607-CD -- ADDITIVE aggregation-phase plumbing on top of the W607-AK
    # substrate-CALL markers. W607-AK already wrapped the substrate-helper
    # boundaries on the build path (find_project_root / discover_and_parse /
    # compute_risk_score / sort_risky_full / top_risky / supply_chain_to_sarif
    # / write_sarif); W607-CD extends marker coverage to the AGGREGATION-PHASE
    # boundaries that W607-AK left unguarded:
    #
    #   - ``compute_predicate``    -- per-field extraction of metrics fields
    #                                 (score, total, direct_count, dev_count,
    #                                 pin_coverage_pct, exact_count,
    #                                 range_count, unpinned_count) used to
    #                                 compose the verdict string + envelope.
    #                                 A future ``compute_risk_score`` schema
    #                                 refactor that drops/renames one of
    #                                 these keys would otherwise crash the
    #                                 envelope post-build.
    #   - ``compute_verdict``      -- verdict string assembly based on score
    #                                 thresholds (>=80 healthy / >=60 fair /
    #                                 <60 risky / empty deps -> no-deps).
    #                                 Floor to a literal "Supply chain
    #                                 analysis completed" string per LAW 6
    #                                 (standalone-parse) + W978 first-
    #                                 hypothesis discipline (no re-
    #                                 interpolation of the same values that
    #                                 just raised).
    #   - ``serialize_envelope``   -- ``json_envelope("supply-chain", ...)``
    #                                 projection (downstream contract changes
    #                                 / shape regressions).
    #
    # cmd_supply_chain is the SBOM/VEX projection leg of the W805 cross-
    # artifact-consistency family. Closes the supply-chain attestation
    # QUARTET together with W607-AD/BT (cmd_attest), W607-AE/BW
    # (cmd_pr_bundle), and W607-AF/BZ (cmd_cga). The W607-CD markers fire
    # AT RUNTIME when an aggregation-phase boundary raises, complementing
    # the W805 xfail-strict pins that catch structural inconsistency at
    # the dataclass level.
    #
    # Marker family ``supply_chain_*`` -- same family as W607-AK (additive,
    # not a separate prefix). Empty bucket -> byte-identical envelope on
    # the success path.
    #
    # No ``auto_log`` phase: cmd_supply_chain has no active-run ledger
    # write at present, so the W607-BZ 4-phase set drops to 3 phases here
    # (compute_predicate / compute_verdict / serialize_envelope). Same
    # marker shape contract, narrower phase set.
    _w607cd_warnings_out: list[str] = []

    def _run_check_cd(phase: str, fn, *args, default=None, **kwargs):
        """Run one aggregation-phase boundary with W607-CD marker emission.

        Mirror of ``_run_check_ak`` shape (same ``supply_chain_<phase>_failed:``
        marker family) but writes into ``_w607cd_warnings_out`` so the
        additive bucket stays distinguishable in tests + audits.
        """
        try:
            return fn(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 -- top-level disclosure
            _w607cd_warnings_out.append(f"supply_chain_{phase}_failed:{type(exc).__name__}:{exc}")
            return default

    project_root = _run_check_ak("find_project_root", find_project_root, default=None)
    if project_root is None:
        project_root = Path.cwd()

    deps = _run_check_ak("discover_and_parse", discover_and_parse, project_root, default=[])
    metrics = _run_check_ak(
        "compute_risk_score",
        compute_risk_score,
        deps,
        default=dict(
            score=0,
            pin_coverage=0.0,
            dev_ratio=0.0,
            total=0,
            direct_count=0,
            dev_count=0,
            exact_count=0,
            range_count=0,
            unpinned_count=0,
        ),
    )
    score = metrics["score"]
    # W1142-followup-B: cap-hit disclosure. ``top_risky`` returns a
    # risk-sorted slice of length <= ``top``; the full pre-slice
    # population is the risk-sorted ``deps`` list. Record the full
    # count so the envelope can disclose when ``--limit`` truncated
    # the riskiest-dependency report.
    risky_full = _run_check_ak(
        "sort_risky_full",
        lambda _deps: sorted(
            _deps,
            key=lambda d: ({"unpinned": 2, "range": 1, "exact": 0}[d.pin_status], not d.is_dev),
            reverse=True,
        ),
        deps,
        default=[],
    )
    total_risky_full = len(risky_full)
    risky = _run_check_ak("top_risky", top_risky, deps, top, default=[])
    risky_truncated = total_risky_full > len(risky)
    found_files = sorted({d.source_file for d in deps})
    ecosystems: dict[str, int] = dict(Counter(d.ecosystem for d in deps))

    # W607-CD -- compute_predicate boundary. Wraps the per-field extraction
    # of metrics so a future ``compute_risk_score`` schema refactor that
    # drops/renames keys surfaces a marker rather than crashing the envelope.
    # Floor to documented empty-shape ints matching the happy-path return so
    # downstream verdict/summary fields stay non-null. Accessors are safe-
    # floored via ``.get()`` so a malformed metrics dict (missing keys)
    # caught by this wrap doesn't trip a KeyError post-disclosure.
    def _compute_predicate_fields(metrics_local: dict) -> dict:
        return {
            "score": metrics_local["score"],
            "pin_coverage": metrics_local["pin_coverage"],
            "unpinned_count": metrics_local["unpinned_count"],
            "range_count": metrics_local["range_count"],
            "exact_count": metrics_local["exact_count"],
            "total": metrics_local["total"],
            "direct_count": metrics_local["direct_count"],
            "dev_count": metrics_local["dev_count"],
        }

    _pred_fields = _run_check_cd(
        "compute_predicate",
        _compute_predicate_fields,
        metrics,
        default={
            "score": 0,
            "pin_coverage": 0.0,
            "unpinned_count": 0,
            "range_count": 0,
            "exact_count": 0,
            "total": 0,
            "direct_count": 0,
            "dev_count": 0,
        },
    )

    # W607-CD -- compute_verdict boundary. Wraps the verdict-string assembly
    # so a downstream f-string refactor (e.g. a non-int pin_coverage from a
    # vocabulary refactor) surfaces a marker rather than crashing the
    # envelope. Floor must NOT re-interpolate the same values that tripped
    # the closure (W978 first-hypothesis discipline: a __format__-raising
    # sentinel under test would re-raise inside the default f-string). Use
    # a literal ``"Supply chain analysis completed"`` floor instead (LAW 6
    # still holds: the line works standalone). Mirror of cmd_cga W607-BZ
    # compute_verdict pattern.
    def _build_verdict_str(fields: dict, has_deps: bool) -> str:
        if not has_deps:
            return "No dependency files found"
        score_local = fields["score"]
        if score_local >= 80:
            _pct_local = int(fields["pin_coverage"] * 100)
            return f"Supply chain healthy ({score_local}/100) -- {_pct_local}% pinned"
        if score_local >= 60:
            _u_local = fields["unpinned_count"]
            _r_local = fields["range_count"]
            return f"Supply chain fair ({score_local}/100) -- {_u_local} unpinned, {_r_local} ranges"
        _u_local = fields["unpinned_count"]
        return f"Supply chain risky ({score_local}/100) -- {_u_local} unpinned dependencies"

    verdict = _run_check_cd(
        "compute_verdict",
        _build_verdict_str,
        _pred_fields,
        bool(deps),
        default="Supply chain analysis completed",
    )

    if sarif_mode:
        sarif = _run_check_ak(
            "supply_chain_to_sarif",
            supply_chain_to_sarif,
            deps,
            score,
            default={},
        )
        _sarif_text = _run_check_ak("write_sarif", write_sarif, sarif, default="")
        click.echo(_sarif_text)
        return

    if json_mode:
        # W1142-followup-B: cap-hit disclosure on the canonical JSON
        # envelope. ``count``/``total_count``/``truncated``/``limit``
        # surface whether the agent's --limit collapsed signal.
        _cap_summary = {
            "count": len(risky),
            "total_count": total_risky_full,
            "truncated": risky_truncated,
            "limit": top,
        }
        _warnings_out: list[str] = []
        if risky_truncated:
            _warnings_out.append(f"truncated to {len(risky)} of {total_risky_full} — pass --limit larger to see more")
        # W607-AK / W607-CD: merge substrate-CALL markers AND aggregation-
        # phase markers into the canonical ``warnings_out`` channel. All
        # three buckets share the canonical ``supply_chain_*`` family
        # (W1142-followup-B uses prose; the W607-* buckets use the
        # ``supply_chain_<phase>_failed:`` shape). The additive bucket
        # stays distinguishable in tests + audits via its phase names
        # (``compute_predicate`` / ``compute_verdict`` /
        # ``serialize_envelope``). ``partial_success`` flips when ANY
        # bucket is non-empty.
        _combined_warnings_out = list(_warnings_out) + list(_w607ak_warnings_out) + list(_w607cd_warnings_out)
        # W607-CD: safe-floor metric accessors via ``.get()`` so a
        # malformed metrics dict (missing keys) caught by the
        # compute_predicate wrap above doesn't trip a KeyError post-
        # disclosure. The W607-CD marker is already on the bucket; the
        # envelope still emits with documented empty-floor values so
        # consumers see the degradation lineage rather than empty stdout.
        _summary = dict(
            verdict=verdict,
            risk_score=_pred_fields.get("score", 0),
            total_dependencies=metrics.get("total", 0),
            direct_count=metrics.get("direct_count", 0),
            dev_count=metrics.get("dev_count", 0),
            pin_coverage_pct=round(_pred_fields.get("pin_coverage", 0.0) * 100, 1)
            if isinstance(_pred_fields.get("pin_coverage"), (int, float))
            else 0.0,
            unpinned_count=metrics.get("unpinned_count", 0),
            range_count=metrics.get("range_count", 0),
            exact_count=metrics.get("exact_count", 0),
            files_scanned=found_files,
            ecosystems=ecosystems,
            **_cap_summary,
        )
        if _combined_warnings_out:
            _summary["warnings_out"] = list(_combined_warnings_out)
            _summary["partial_success"] = True
        _envelope_kwargs: dict = dict(
            budget=token_budget,
            risk_score=_pred_fields.get("score", 0),
            total_dependencies=metrics.get("total", 0),
            direct_count=metrics.get("direct_count", 0),
            dev_count=metrics.get("dev_count", 0),
            exact_count=metrics.get("exact_count", 0),
            range_count=metrics.get("range_count", 0),
            unpinned_count=metrics.get("unpinned_count", 0),
            pin_coverage_pct=round(_pred_fields.get("pin_coverage", 0.0) * 100, 1)
            if isinstance(_pred_fields.get("pin_coverage"), (int, float))
            else 0.0,
            files_scanned=found_files,
            ecosystems=ecosystems,
            top_risky=[
                dict(
                    name=d.name,
                    version_spec=d.version_spec,
                    pin_status=d.pin_status,
                    risk_level=d.risk_level,
                    is_dev=d.is_dev,
                    ecosystem=d.ecosystem,
                    source_file=d.source_file,
                )
                for d in risky
            ],
            all_dependencies=[
                dict(
                    name=d.name,
                    version_spec=d.version_spec,
                    pin_status=d.pin_status,
                    risk_level=d.risk_level,
                    is_dev=d.is_dev,
                    ecosystem=d.ecosystem,
                    source_file=d.source_file,
                )
                for d in deps
            ],
        )
        if _combined_warnings_out:
            _envelope_kwargs["warnings_out"] = list(_combined_warnings_out)
            _envelope_kwargs["partial_success"] = True

        # W607-CD -- serialize_envelope boundary. Wraps the envelope
        # serialization itself. A downstream schema-shape refactor that
        # breaks ``json_envelope("supply-chain", ...)`` would otherwise
        # crash AFTER all substrate + aggregation signals were already
        # gathered. Floor to a minimal envelope stub so consumers still
        # receive a parseable JSON object with the marker attached + the
        # canonical command name. Mirror of cmd_cga's W607-BZ
        # serialize_envelope floor pattern.
        _envelope_floor: dict = {
            "command": "supply-chain",
            "schema_version": "1.0.0",
            "summary": {
                "verdict": verdict,
                "partial_success": True,
                "warnings_out": list(_combined_warnings_out),
            },
            "warnings_out": list(_combined_warnings_out),
        }
        envelope = _run_check_cd(
            "serialize_envelope",
            json_envelope,
            "supply-chain",
            default=_envelope_floor,
            summary=_summary,
            **_envelope_kwargs,
        )
        # W607-CD -- if ``serialize_envelope`` raised AFTER the combined
        # bucket was already snapshotted, the new
        # ``supply_chain_serialize_envelope_failed:`` marker was appended
        # to ``_w607cd_warnings_out`` and the floor stub carries only the
        # pre-raise combined list. Rebuild the floor stub's warnings_out
        # so the new marker reaches the JSON output. Clean path ->
        # envelope is the real json_envelope return value, no rebuild
        # needed.
        if envelope is _envelope_floor and _w607cd_warnings_out:
            _combined_warnings_out = list(_warnings_out) + list(_w607ak_warnings_out) + list(_w607cd_warnings_out)
            _envelope_floor["summary"]["warnings_out"] = list(_combined_warnings_out)
            _envelope_floor["warnings_out"] = list(_combined_warnings_out)
            envelope = _envelope_floor

        click.echo(to_json(envelope))
        return

    click.echo(f"VERDICT: {verdict}")
    click.echo()

    if not deps:
        click.echo("  No dependency files found.")
        return

    _t = metrics["total"]
    _dc = metrics["direct_count"]
    _dv = metrics["dev_count"]
    _pct = int(metrics["pin_coverage"] * 100)
    _ex = metrics["exact_count"]
    _rng = metrics["range_count"]
    _unp = metrics["unpinned_count"]
    click.echo(f"Risk Score: {score}/100  |  Total: {_t}  |  Direct: {_dc}  |  Dev: {_dv}")
    click.echo(f"Pin Coverage: {_pct}%  |  Exact: {_ex}  |  Range: {_rng}  |  Unpinned: {_unp}")
    if found_files:
        click.echo("Files: " + ", ".join(found_files))
    if ecosystems:
        eco_str = "  ".join(f"{k}:{v}" for k, v in sorted(ecosystems.items()))
        click.echo(f"Ecosystems: {eco_str}")
    click.echo()

    if risky:
        _topn = min(top, len(risky))
        click.echo(f"=== Top {_topn} Riskiest Dependencies ===")
        rows = []
        for d in risky:
            spec = d.version_spec if d.version_spec else "(none)"
            dev_str = "dev" if d.is_dev else "direct"
            rows.append([d.name, spec, d.pin_status, d.risk_level, dev_str, d.ecosystem, d.source_file])
        click.echo(
            format_table(
                ["Name", "Version Spec", "Pin Status", "Risk", "Type", "Ecosystem", "File"],
                rows,
            )
        )
        click.echo()

    recs = []
    if metrics["unpinned_count"] > 0:
        _u = metrics["unpinned_count"]
        recs.append(f"Pin {_u} unpinned dependencies to exact versions")
    if metrics["range_count"] > 5:
        _r = metrics["range_count"]
        recs.append(f"Consider pinning {_r} range-versioned dependencies")
    if not recs:
        recs.append("Dependency pinning looks good")
    click.echo("Recommendations:")
    for rec in recs:
        click.echo(f"  - {rec}")
