#!/usr/bin/env python3
"""Build a deterministic verification plan for execution by AI-Collab.

PRD Plugin owns policy, exact changed-file discovery, and conservative fallback
rules. AI-Collab owns impact lookup, structural expansion, test mapping, and
test execution. This module is read-only and never requires a git remote.
"""

from __future__ import annotations

import argparse
import fnmatch
import hashlib
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any, Iterable

import prd_config
import prd_substrate


SCHEMA_VERSION = "1.0"
CONTRACT_VERSION = 1


class PlanError(ValueError):
    """The verification policy or local repository state is invalid."""


def _canonical_hash(value: Any) -> str:
    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _normalize_path(value: str) -> str:
    path = str(value).strip().replace("\\", "/")
    while path.startswith("./"):
        path = path[2:]
    parts = [part for part in path.split("/") if part not in ("", ".")]
    if (
        not parts
        or path.startswith("/")
        or re.match(r"^[A-Za-z]:/", path)
        or Path(path).is_absolute()
        or any(part == ".." for part in parts)
    ):
        raise PlanError(f"changed path must be repo-relative: {value!r}")
    return "/".join(parts)


def _sorted_paths(values: Iterable[str]) -> list[str]:
    normalized = {_normalize_path(value) for value in values}
    return sorted(normalized, key=lambda value: (value.lower(), value))


def _git_paths(root: Path, *args: str) -> list[str]:
    completed = subprocess.run(
        ["git", *args],
        cwd=root,
        capture_output=True,
        check=False,
    )
    if completed.returncode != 0:
        message = completed.stderr.decode("utf-8", errors="replace").strip()
        raise PlanError(message or f"git {' '.join(args)} failed")
    return [part.decode("utf-8", errors="surrogateescape") for part in completed.stdout.split(b"\0") if part]


def collect_changed_files(repo_root: str | Path = ".", *, base_ref: str = "") -> list[str]:
    """Return tracked and untracked local changes, or a supplied commit range.

    The default compares the working tree/index with HEAD and includes untracked
    non-ignored files. ``base_ref`` adds committed changes from ``base_ref...HEAD``
    while still including current local changes. No remote lookup is performed.
    """
    root = Path(repo_root).resolve()
    try:
        top = _git_paths(root, "rev-parse", "--show-toplevel")
    except PlanError as exc:
        raise PlanError(f"repository root is not a git worktree: {root}: {exc}") from exc
    if not top or Path(top[0].strip()).resolve() != root:
        raise PlanError(f"repo_root must be the git worktree root: {root}")

    changed: list[str] = []
    if base_ref:
        changed.extend(_git_paths(root, "diff", "--name-only", "-z", f"{base_ref}...HEAD"))
    try:
        changed.extend(_git_paths(root, "diff", "--name-only", "-z", "HEAD"))
    except PlanError:
        changed.extend(_git_paths(root, "diff", "--name-only", "-z", "--cached"))
        changed.extend(_git_paths(root, "diff", "--name-only", "-z"))
    changed.extend(_git_paths(root, "ls-files", "--others", "--exclude-standard", "-z"))
    return _sorted_paths(changed)


def _policy(root: Path) -> dict[str, Any]:
    def get(key: str) -> Any:
        return prd_config.get(root, key)

    policy = {
        "enabled": get("verification.test_scope.enabled"),
        "executor": get("verification.test_scope.executor"),
        "neighbor_limit": get("verification.test_scope.neighbor_limit"),
        "max_changed_files": get("verification.test_scope.max_changed_files"),
        "fallback": get("verification.test_scope.fallback"),
        "full_suite_triggers": get("verification.test_scope.full_suite_triggers"),
        "core_paths": get("verification.test_scope.core_paths"),
        "test_patterns": get("verification.test_scope.test_patterns"),
        "max_selected_tests": get("verification.test_scope.max_selected_tests"),
        "execution_timeout_seconds": get("verification.test_scope.execution_timeout_seconds"),
        "focused_commands": get("verification.test_scope.focused_commands"),
        "full_commands": get("verification.test_scope.full_commands"),
        "allowed_executables": get("verification.test_scope.allowed_executables"),
    }
    if not isinstance(policy["enabled"], bool):
        raise PlanError("verification.test_scope.enabled must be boolean")
    if policy["executor"] != "ai-collab":
        raise PlanError("verification.test_scope.executor must be 'ai-collab'")
    if policy["fallback"] != "full":
        raise PlanError("verification.test_scope.fallback must be 'full'")
    for key, lower, upper in (("neighbor_limit", 1, 100), ("max_changed_files", 1, 1000),
                              ("max_selected_tests", 1, 5000), ("execution_timeout_seconds", 1, 86400)):
        value = policy[key]
        if isinstance(value, bool) or not isinstance(value, int) or not lower <= value <= upper:
            raise PlanError(f"verification.test_scope.{key} must be {lower}..{upper}")
    allowed_triggers = set(prd_config.VERIFICATION_FULL_SUITE_TRIGGERS)
    for key in ("full_suite_triggers", "core_paths", "test_patterns", "allowed_executables"):
        value = policy[key]
        if not isinstance(value, list) or any(not isinstance(item, str) or not item for item in value):
            raise PlanError(f"verification.test_scope.{key} must be a list of non-empty strings")
        if len(value) != len(set(value)):
            raise PlanError(f"verification.test_scope.{key} must not contain duplicates")
    unknown = [item for item in policy["full_suite_triggers"] if item not in allowed_triggers]
    if unknown:
        raise PlanError(f"verification.test_scope.full_suite_triggers contains unknown values: {unknown}")
    missing = [
        item for item in prd_config.VERIFICATION_FULL_SUITE_TRIGGERS
        if item not in policy["full_suite_triggers"]
    ]
    if missing:
        raise PlanError(
            f"verification.test_scope.full_suite_triggers is missing mandatory values: {missing}"
        )
    return policy


def _matches_any(path: str, patterns: Iterable[str]) -> bool:
    return any(fnmatch.fnmatchcase(path, pattern.replace("\\", "/")) for pattern in patterns)


def _with_hash(plan: dict[str, Any]) -> dict[str, Any]:
    plan["plan_hash"] = _canonical_hash(plan)
    return plan


def build_plan(
    repo_root: str | Path = ".",
    *,
    changed_files: Iterable[str] | None = None,
    base_ref: str = "",
    release: bool = False,
) -> dict[str, Any]:
    """Build a portable plan; never call AI-Collab or execute tests."""
    root = Path(repo_root).resolve()
    files = collect_changed_files(root, base_ref=base_ref) if changed_files is None else _sorted_paths(changed_files)
    policy = _policy(root)
    common = {
        "schema_version": SCHEMA_VERSION,
        "contract_version": CONTRACT_VERSION,
        "executor": policy["executor"],
        "changed_files": files,
        "change_fingerprint": _canonical_hash(files),
        "policy": policy,
        "selection_rules": {
            "precedence": [
                "exact_changed",
                "explicit_test_ownership",
                "structural_dependencies",
                "predictive_impact",
            ],
            "require_all_changed_files_mapped": True,
            "widen_before_full": True,
            "predictive_signal_is_sufficient_alone": False,
        },
        "fallback": {
            "strategy": "full",
            "conditions": list(policy["full_suite_triggers"]),
        },
    }

    reason = ""
    if release:
        reason = "release"
    elif not files:
        return _with_hash({**common, "strategy": "none", "reason": "no_changes", "runtime_requests": {}})
    elif not policy["enabled"]:
        reason = "test_scope_disabled"
    elif any(_matches_any(path, policy["core_paths"]) for path in files):
        reason = "core_change"
    elif len(files) > policy["max_changed_files"]:
        reason = "max_changed_files_exceeded"
    else:
        substrate = prd_substrate.effective_policy(root)
        if (substrate["effective_mode"] != "coordinate"
                or "verification" not in substrate["effective_capabilities"]
                or not substrate["configured"]["automation"]["verification_execution"]):
            reason = "executor_unavailable"

    if reason:
        return _with_hash({**common, "strategy": "full", "reason": reason, "runtime_requests": {}})

    runtime_requests = {
        "impact": [
            {"path": path, "query": path, "k": policy["neighbor_limit"]}
            for path in files
        ],
        "structural": {
            "subjects": files,
            "callers": True,
            "references": True,
        },
        "test_mapping": {
            "subjects": files,
            "patterns": list(policy["test_patterns"]),
            "use_explicit_ownership": True,
            "require_all_changed_files_mapped": True,
        },
    }
    return _with_hash({**common, "strategy": "impact_scoped", "runtime_requests": runtime_requests})


def _validate_plan(plan: dict[str, Any]) -> None:
    if not isinstance(plan, dict) or plan.get("contract_version") != CONTRACT_VERSION:
        raise PlanError("verification plan has an unsupported contract")
    supplied = plan.get("plan_hash")
    unhashed = {key: value for key, value in plan.items() if key != "plan_hash"}
    if supplied != _canonical_hash(unhashed):
        raise PlanError("verification plan hash does not match its contents")
    if plan.get("change_fingerprint") != _canonical_hash(plan.get("changed_files", [])):
        raise PlanError("verification change fingerprint does not match changed_files")


def _repo_files(root: Path) -> list[str]:
    rows = []
    for path in root.rglob("*"):
        if not path.is_file():
            continue
        rel = path.relative_to(root).as_posix()
        if rel.startswith((".git/", ".prd_plugin/local/")):
            continue
        rows.append(rel)
    return sorted(set(rows), key=lambda value: (value.lower(), value))


def _runtime_paths(value: Any, root: Path) -> set[str]:
    found: set[str] = set()
    if isinstance(value, dict):
        for key, child in value.items():
            if key in {"path", "file", "filename", "source"} and isinstance(child, str):
                try:
                    normalized = _normalize_path(child)
                    if (root / normalized).is_file():
                        found.add(normalized)
                except PlanError:
                    pass
            found.update(_runtime_paths(child, root))
    elif isinstance(value, list):
        for child in value:
            found.update(_runtime_paths(child, root))
    return found


def _map_tests(root: Path, subjects: set[str], policy: dict[str, Any]) -> tuple[list[str], list[str]]:
    files = _repo_files(root)
    tests = [path for path in files if _matches_any(path, policy["test_patterns"])]
    selected = {path for path in subjects if path in tests}
    unmapped = []
    for subject in sorted(subjects):
        if subject in tests:
            continue
        suffix = Path(subject).suffix.lower()
        if suffix not in {".py", ".js", ".jsx", ".ts", ".tsx", ".go", ".rs", ".java", ".cs"}:
            continue
        stem = Path(subject).stem.lower()
        candidates = [path for path in tests if stem and stem in Path(path).stem.lower()]
        if candidates:
            selected.update(candidates)
        else:
            unmapped.append(subject)
    return sorted(selected), unmapped


def _configured_commands(raw: Any, tests: list[str]) -> list[list[str]]:
    if not raw:
        return []
    if not isinstance(raw, list) or any(not isinstance(row, list) or not row for row in raw):
        raise PlanError("verification commands must be arrays of non-empty argv arrays")
    commands = []
    for row in raw:
        if any(not isinstance(item, str) or not item for item in row):
            raise PlanError("verification command arguments must be non-empty strings")
        expanded = []
        for item in row:
            if item == "{tests}":
                expanded.extend(tests)
            else:
                expanded.append(item)
        commands.append(expanded)
    return commands


def _auto_commands(root: Path, strategy: str, tests: list[str]) -> list[list[str]]:
    if strategy == "impact_scoped" and tests:
        suffixes = {Path(path).suffix.lower() for path in tests}
        if suffixes == {".py"}:
            return [[sys.executable, "-m", "unittest", *tests]]
        if suffixes <= {".js", ".jsx", ".ts", ".tsx"} and (root / "package.json").is_file():
            return [["npm", "test", "--", *tests]]
        if suffixes == {".go"}:
            packages = sorted({"./" + str(Path(path).parent).replace("\\", "/") for path in tests})
            return [["go", "test", *packages]]
    if (root / "package.json").is_file():
        try:
            package = json.loads((root / "package.json").read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError):
            package = {}
        if isinstance(package.get("scripts"), dict) and package["scripts"].get("test"):
            return [["npm", "test"]]
    if (root / "tests").is_dir() and any((root / "tests").rglob("test_*.py")):
        return [[sys.executable, "-m", "unittest", "discover", "-s", "tests", "-q"]]
    if (root / "go.mod").is_file():
        return [["go", "test", "./..."]]
    if (root / "Cargo.toml").is_file():
        return [["cargo", "test"]]
    raise PlanError("no safe verification command could be detected; configure verification.test_scope commands")


def _run_commands(root: Path, commands: list[list[str]], policy: dict[str, Any]) -> list[dict[str, Any]]:
    allowed = {name.lower().removesuffix(".exe").removesuffix(".cmd") for name in policy["allowed_executables"]}
    receipts = []
    for command in commands:
        executable = Path(command[0]).name.lower().removesuffix(".exe").removesuffix(".cmd")
        if executable not in allowed:
            raise PlanError(f"verification executable is not allowlisted: {command[0]}")
        # REQ-166: Windows command shims such as npm.cmd are not reliably
        # CreateProcess-resolvable from their bare token when shell=False.
        resolved_executable = shutil.which(command[0])
        if not resolved_executable:
            raise PlanError(f"verification executable was not found: {command[0]}")
        resolved_command = [resolved_executable, *command[1:]]
        try:
            completed = subprocess.run(
                resolved_command, cwd=root, text=True, encoding="utf-8", errors="replace",
                capture_output=True, check=False, timeout=policy["execution_timeout_seconds"],
            )
        except subprocess.TimeoutExpired as exc:
            raise PlanError(f"verification command timed out: {command[0]}") from exc
        receipt_command = ["python" if item == sys.executable else item for item in command]
        receipt = {
            "command": receipt_command, "exit_code": completed.returncode,
            "stdout": completed.stdout[-20000:], "stderr": completed.stderr[-20000:],
        }
        receipt["receipt_hash"] = _canonical_hash(receipt)
        receipts.append(receipt)
        if completed.returncode:
            break
    return receipts


def execute_plan(repo_root: str | Path, plan: dict[str, Any] | None = None) -> dict[str, Any]:
    """Execute a hash-bound plan, widening conservatively whenever scope is uncertain."""
    root = Path(repo_root).resolve()
    plan = build_plan(root) if plan is None else plan
    _validate_plan(plan)
    policy = _policy(root)
    strategy = plan["strategy"]
    reason = plan.get("reason", "")
    selected: list[str] = []
    impact_receipts = []
    if strategy == "none":
        return {"status": "passed", "strategy": "none", "reason": reason, "selected_tests": [],
                "change_fingerprint": plan["change_fingerprint"], "plan_hash": plan["plan_hash"], "receipts": []}
    if strategy == "impact_scoped":
        subjects = set(plan["changed_files"])
        for request in plan["runtime_requests"]["impact"]:
            try:
                outcome = prd_substrate.execute_tool(
                    root, tool="impact", arguments={**request, "repo": root.name}, capability="impact",
                    source_refs=[request["path"]], idempotency_key=f"prd-impact:{plan['change_fingerprint']}:{request['path']}",
                )
            except prd_substrate.ContractError as exc:
                outcome = {"status": "degraded", "reason": str(exc)}
            impact_receipts.append(outcome)
            if outcome.get("status") != "completed":
                strategy, reason = "full", "impact_degraded"
                break
            subjects.update(_runtime_paths(outcome.get("receipt", {}).get("result"), root))
        if strategy == "impact_scoped":
            selected, unmapped = _map_tests(root, subjects, policy)
            if unmapped or not selected:
                strategy, reason = "full", "unmapped_change"
            elif len(selected) > policy["max_selected_tests"]:
                strategy, reason = "full", "selected_test_limit"
    raw = policy["focused_commands"] if strategy == "impact_scoped" else policy["full_commands"]
    commands = _configured_commands(raw, selected)
    if not commands:
        commands = _auto_commands(root, strategy, selected)
    receipts = _run_commands(root, commands, policy)
    passed = bool(receipts) and all(row["exit_code"] == 0 for row in receipts)
    result = {
        "status": "passed" if passed else "failed", "strategy": strategy, "reason": reason,
        "selected_tests": selected if strategy == "impact_scoped" else [],
        "change_fingerprint": plan["change_fingerprint"], "plan_hash": plan["plan_hash"],
        "impact_receipts": impact_receipts, "receipts": receipts,
    }
    result["execution_hash"] = _canonical_hash(result)
    return result


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--repo-root", default=".")
    parser.add_argument("--base-ref", default="")
    parser.add_argument("--changed-file", action="append", default=None)
    parser.add_argument("--release", action="store_true")
    parser.add_argument("--execute", action="store_true")
    parser.add_argument("--format", choices=("json",), default="json")
    args = parser.parse_args(argv)
    try:
        plan = build_plan(
            args.repo_root,
            changed_files=args.changed_file,
            base_ref=args.base_ref,
            release=args.release,
        )
        result = execute_plan(args.repo_root, plan) if args.execute else plan
    except (PlanError, prd_substrate.ContractError) as exc:
        print(f"[PRD Plugin] verification error: {exc}", file=sys.stderr)
        return 2
    print(json.dumps(result, indent=2))
    return 0 if result.get("status") != "failed" else 1


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