"""Enforce focused KnowledgeFS coverage without hiding integration-critical glue.

The primary threshold aggregates statement and branch coverage for every Dify
module owned by the KnowledgeFS integration. Large pre-existing Dify modules
that only contain narrow integration hooks are checked with changed-line
coverage instead, so unrelated legacy code cannot dilute or inflate the gate.
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import TypedDict, cast

WORKSPACE_ROOT = Path(__file__).resolve().parents[2]
NON_CORE_COVERAGE_ALLOWLIST = frozenset(
    {
        "api/dev/check_knowledge_fs_coverage.py",
        "api/dev/generate_knowledge_fs_contract.py",
        "api/dev/knowledge_fs_product_contract.py",
        "api/migrations/versions/2026_07_21_1200-a4e7c2f91b30_add_knowledge_fs_control_plane.py",
        "api/migrations/versions/2026_07_21_1300-b7f2a9d41c60_add_knowledge_fs_cutover.py",
        "api/migrations/versions/2026_07_21_1400-c8e31b7d52a4_add_knowledge_fs_cleanup_authorization.py",
    }
)
HUNK_HEADER = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")
logger = logging.getLogger(__name__)


class CoverageSummary(TypedDict):
    """Coverage.py counts required by the aggregate gate."""

    covered_lines: int
    num_statements: int
    covered_branches: int
    num_branches: int


class CoverageFile(TypedDict):
    """Per-file coverage data emitted by ``coverage json``."""

    executed_lines: list[int]
    missing_lines: list[int]
    summary: CoverageSummary


class CoverageReport(TypedDict):
    """Relevant top-level shape of a coverage.py JSON report."""

    files: dict[str, CoverageFile]


@dataclass(frozen=True, slots=True)
class CoverageTotals:
    """Covered and measurable units for one gate surface."""

    covered: int
    total: int

    @property
    def percent(self) -> float:
        return 100.0 if self.total == 0 else self.covered * 100 / self.total


class CoverageGateError(RuntimeError):
    """Raised when coverage input is incomplete or below its threshold."""


def main() -> None:
    """Validate focused module coverage and changed integration glue."""
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    parser = argparse.ArgumentParser()
    parser.add_argument("--coverage-json", type=Path, required=True)
    parser.add_argument("--glue-manifest", type=Path, required=True)
    parser.add_argument("--workspace-root", type=Path, default=WORKSPACE_ROOT)
    parser.add_argument("--base", default="")
    parser.add_argument("--minimum", type=float, default=90.0)
    parser.add_argument("--glue-minimum", type=float, default=90.0)
    args = parser.parse_args()

    workspace_root = args.workspace_root.resolve()
    report = load_coverage_report(args.coverage_json)
    core_totals = validate_core_coverage(report, workspace_root=workspace_root, minimum=args.minimum)
    base = resolve_diff_base(workspace_root, args.base)
    glue_paths = load_glue_coverage_paths(args.glue_manifest, workspace_root=workspace_root)
    changed_lines = collect_changed_glue_lines(workspace_root, base, glue_paths=glue_paths)
    glue_totals = validate_changed_glue_coverage(
        report,
        changed_lines=changed_lines,
        minimum=args.glue_minimum,
    )
    logger.info(
        "Dify KnowledgeFS coverage passed: core lines+branches %.2f%% (%d/%d); changed glue lines %.2f%% (%d/%d)",
        core_totals.percent,
        core_totals.covered,
        core_totals.total,
        glue_totals.percent,
        glue_totals.covered,
        glue_totals.total,
    )


def load_coverage_report(path: Path) -> CoverageReport:
    """Load the detailed JSON report used by both coverage checks."""
    if not path.is_file():
        raise CoverageGateError(f"coverage JSON does not exist: {path}")
    document = json.loads(path.read_text())
    if not isinstance(document, dict) or not isinstance(document.get("files"), dict):
        raise CoverageGateError(f"coverage JSON has no files object: {path}")
    return cast(CoverageReport, document)


def load_glue_coverage_paths(path: Path, *, workspace_root: Path) -> tuple[str, ...]:
    """Load the workflow's authoritative NUL-delimited integration touchpoints."""
    if not path.is_file():
        raise CoverageGateError(f"KnowledgeFS glue manifest does not exist: {path}")
    try:
        paths = tuple(item.decode() for item in path.read_bytes().split(b"\0") if item)
    except UnicodeDecodeError as error:
        raise CoverageGateError(f"KnowledgeFS glue manifest is not UTF-8: {path}") from error
    if not paths:
        raise CoverageGateError("Dify KnowledgeFS glue coverage target set is empty")
    if len(paths) != len(set(paths)):
        raise CoverageGateError("Dify KnowledgeFS glue coverage manifest contains duplicate paths")
    invalid_paths = [
        candidate
        for candidate in paths
        if not candidate.startswith("api/")
        or not candidate.endswith(".py")
        or not (workspace_root / candidate).is_file()
    ]
    if invalid_paths:
        raise CoverageGateError(f"Dify KnowledgeFS glue coverage paths are invalid: {', '.join(invalid_paths)}")
    return paths


def is_core_coverage_path(path: str) -> bool:
    """Return whether a repository-relative path belongs to the focused aggregate."""
    if not path.endswith(".py"):
        return False
    if path in {
        "api/commands/knowledge_fs.py",
        "api/configs/extra/knowledge_fs_config.py",
        "api/extensions/ext_knowledge_fs_observability.py",
        "api/services/knowledge_fs_capability.py",
    }:
        return True
    if path.startswith(
        (
            "api/controllers/console/knowledge_fs/",
            "api/controllers/service_api/knowledge_fs/",
            "api/core/tools/builtin_tool/providers/knowledge_fs/",
            "api/services/knowledge_fs/",
        )
    ):
        return True
    filename = path.rsplit("/", maxsplit=1)[-1]
    return (
        path.startswith("api/models/")
        and filename.startswith("knowledge_fs")
        or path.startswith("api/repositories/")
        and "knowledge_fs" in filename
        or path.startswith("api/tasks/")
        and "knowledge_fs" in filename
    )


def discover_core_coverage_paths(workspace_root: Path) -> tuple[str, ...]:
    """Classify every KnowledgeFS-named production file or fail closed."""
    named_paths = discover_knowledge_fs_production_paths(workspace_root)
    core_paths = {path for path in named_paths if is_core_coverage_path(path)}
    unclassified_paths = set(named_paths) - core_paths - NON_CORE_COVERAGE_ALLOWLIST
    if unclassified_paths:
        raise CoverageGateError(
            "unclassified Dify KnowledgeFS production files must join the core coverage scope or explicit allowlist: "
            + ", ".join(sorted(unclassified_paths))
        )
    if not core_paths:
        raise CoverageGateError("Dify KnowledgeFS core coverage target set is empty")
    return tuple(sorted(core_paths))


def discover_knowledge_fs_production_paths(workspace_root: Path) -> tuple[str, ...]:
    """Mirror the workflow's dynamic KnowledgeFS filename discovery."""
    api_root = workspace_root / "api"
    if not api_root.is_dir():
        raise CoverageGateError(f"Dify API directory does not exist: {api_root}")
    paths: set[str] = set()
    for directory, child_directories, filenames in os.walk(api_root):
        current_directory = Path(directory)
        if current_directory == api_root:
            child_directories[:] = [name for name in child_directories if name not in {".venv", "storage", "tests"}]
        child_directories[:] = [name for name in child_directories if name != "__pycache__"]
        for filename in filenames:
            path = (current_directory / filename).relative_to(workspace_root).as_posix()
            if filename.endswith(".py") and "knowledge_fs" in path:
                paths.add(path)
    if not paths:
        raise CoverageGateError("Dify KnowledgeFS production target set is empty")
    return tuple(sorted(paths))


def validate_core_coverage(
    report: CoverageReport,
    *,
    workspace_root: Path,
    minimum: float,
) -> CoverageTotals:
    """Require the exact combined line-and-branch percentage for all core files."""
    paths = discover_core_coverage_paths(workspace_root)
    missing_paths = [path for path in paths if path not in report["files"]]
    if missing_paths:
        raise CoverageGateError(f"coverage report is missing core files: {', '.join(missing_paths)}")

    covered = 0
    total = 0
    for path in paths:
        summary = report["files"][path]["summary"]
        covered += summary["covered_lines"] + summary["covered_branches"]
        total += summary["num_statements"] + summary["num_branches"]
    if total == 0:
        raise CoverageGateError("Dify KnowledgeFS core coverage has no measurable statements or branches")

    totals = CoverageTotals(covered=covered, total=total)
    _require_minimum(totals, minimum=minimum, label="Dify KnowledgeFS core line-and-branch coverage")
    return totals


def resolve_diff_base(workspace_root: Path, preferred: str) -> str:
    """Resolve an explicit event base, falling back to the previous commit for manual runs."""
    base = preferred.strip()
    if not base or set(base) == {"0"}:
        base = "HEAD^"
    result = subprocess.run(
        ["git", "cat-file", "-e", f"{base}^{{commit}}"],
        cwd=workspace_root,
        check=False,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        detail = result.stderr.strip() or "commit is unavailable"
        raise CoverageGateError(f"cannot resolve coverage diff base {base}: {detail}")
    return base


def collect_changed_glue_lines(
    workspace_root: Path,
    base: str,
    *,
    glue_paths: tuple[str, ...],
) -> dict[str, set[int]]:
    """Return added line numbers in the narrow Dify modules touched by this integration."""
    result = subprocess.run(
        [
            "git",
            "diff",
            "--no-ext-diff",
            "--no-color",
            "--unified=0",
            base,
            "--",
            *glue_paths,
        ],
        cwd=workspace_root,
        check=False,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        detail = result.stderr.strip() or "git diff failed"
        raise CoverageGateError(f"cannot collect KnowledgeFS glue diff from {base}: {detail}")
    return parse_added_lines(result.stdout)


def parse_added_lines(diff: str) -> dict[str, set[int]]:
    """Parse repository paths and added-side line numbers from a zero-context Git diff."""
    changed_lines: dict[str, set[int]] = {}
    current_path: str | None = None
    current_line: int | None = None
    for raw_line in diff.splitlines():
        if raw_line.startswith("diff --git "):
            current_line = None
            continue
        if raw_line.startswith("+++ "):
            candidate = raw_line[4:]
            current_path = candidate[2:] if candidate.startswith("b/") else None
            if current_path is not None:
                changed_lines.setdefault(current_path, set())
            current_line = None
            continue
        if raw_line.startswith("@@ "):
            match = HUNK_HEADER.match(raw_line)
            current_line = int(match.group(1)) if match is not None else None
            continue
        if current_path is None or current_line is None:
            continue
        if raw_line.startswith("+"):
            changed_lines[current_path].add(current_line)
            current_line += 1
        elif raw_line.startswith("-") or raw_line.startswith("\\"):
            continue
        else:
            current_line += 1
    return changed_lines


def validate_changed_glue_coverage(
    report: CoverageReport,
    *,
    changed_lines: dict[str, set[int]],
    minimum: float,
) -> CoverageTotals:
    """Require added executable glue lines to be exercised by the focused unit suite."""
    covered = 0
    total = 0
    for path, lines in sorted(changed_lines.items()):
        if not lines:
            continue
        file_coverage = report["files"].get(path)
        if file_coverage is None:
            raise CoverageGateError(f"coverage report is missing changed glue file: {path}")
        executed_lines = set(file_coverage["executed_lines"])
        executable_lines = executed_lines | set(file_coverage["missing_lines"])
        changed_executable_lines = lines & executable_lines
        covered += len(changed_executable_lines & executed_lines)
        total += len(changed_executable_lines)

    totals = CoverageTotals(covered=covered, total=total)
    _require_minimum(totals, minimum=minimum, label="Dify KnowledgeFS changed-glue line coverage")
    return totals


def _require_minimum(totals: CoverageTotals, *, minimum: float, label: str) -> None:
    if totals.percent + 1e-12 < minimum:
        raise CoverageGateError(f"{label} is {totals.percent:.2f}%; minimum {minimum:.2f}%")


if __name__ == "__main__":
    main()