PlantAI-rag/tools/audit_migration_requirements.py

#!/usr/bin/env python3
"""Read-only migration requirement audit for PlantAI-rag."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable, Sequence


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from tools.audit_functional_surface_parity import (  # noqa: E402
    build_report as build_functional_parity_report,
)
from tools.audit_functional_surface_parity import resolve_source_root  # noqa: E402
from tools.audit_open_source_readiness import audit_files, build_report as build_open_source_report  # noqa: E402
from tools.runtime_surface import required_tracked_files  # noqa: E402


REQUIRED_TRACKED_FILES = set(required_tracked_files())
EXTERNAL_RESOURCES_MANIFEST = Path("config/external_resources_manifest.json")

REQUIRED_ENV_KEYS = {
    "PC_DATA_ROOT",
    "PLANTAI_SPLIT_DATA_ROOT",
    "PLANTAI_SOURCE_ROOT",
    "BBAV5_ROOT",
    "PLANTAI_CONDA_ROOT",
    "PLANTAI_CONDA_PROFILE",
    "MINERU_CMD",
    "DEEPSEEK_API_KEY",
    "DASHSCOPE_API_KEY",
    "NEO4J_PASSWORD",
}

PRIVATE_ENV_TEMPLATE_KEYS = {
    "BBAV5_REAL_JOB_ID",
    "DEEPSEEK_API_KEY",
    "DASHSCOPE_API_KEY",
    "GF_PYTHON",
    "GF_SECURITY_ADMIN_PASSWORD",
    "MINERU_CMD",
    "NEO4J_PASSWORD",
    "PLANTAI_CONDA_PROFILE",
    "PLANTAI_CONDA_ROOT",
    "PLANTAI_PROJECT_ROOT",
    "PLANTAI_PYTHON",
    "PLANTAI_TEST_PDF",
}

FORBIDDEN_TRACKED_PREFIXES = (
    "data/",
    "paper/",
    "output/",
    "experiments/",
    "mineru/",
    "bge-reranker-base/",
    "v14_learning/",
    "v15_output/",
)

FORBIDDEN_TRACKED_SUFFIXES = (
    ".pdf",
    ".xlsx",
    ".xls",
    ".docx",
    ".pptx",
    ".safetensors",
    ".tar.gz",
    ".zip",
)

FORBIDDEN_PRIVATE_TRACKED_EXACT = (
    ".env",
    ".env.local",
    ".envrc",
)

FORBIDDEN_PRIVATE_TRACKED_PREFIXES = (
    ".agents/",
    ".codegraph/",
    ".codex/",
)


@dataclass(frozen=True)
class RequirementCheck:
    name: str
    status: str
    message: str
    evidence: object


def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Audit migration constraints without editing code or data."
    )
    parser.add_argument("--root", type=Path, default=PROJECT_ROOT, help="PlantAI-rag root.")
    parser.add_argument(
        "--source-root",
        type=Path,
        default=None,
        help="Read-only source project root.",
    )
    parser.add_argument("--json", action="store_true", help="Emit JSON instead of text.")
    parser.add_argument("--no-fail", action="store_true", help="Always exit 0.")
    return parser.parse_args(list(argv) if argv is not None else None)


def git_tracked_files(root: Path) -> list[str]:
    result = subprocess.run(
        ["git", "-C", str(root), "ls-files", "-z"],
        check=True,
        capture_output=True,
    )
    return [item.decode("utf-8") for item in result.stdout.split(b"\0") if item]


def git_remotes(root: Path) -> list[str]:
    result = subprocess.run(
        ["git", "-C", str(root), "remote", "-v"],
        check=True,
        capture_output=True,
        text=True,
    )
    return [line for line in result.stdout.splitlines() if line.strip()]


def tracked_forbidden_data_files(tracked_files: Sequence[str]) -> list[str]:
    offenders = []
    for path in tracked_files:
        if path.startswith(FORBIDDEN_TRACKED_PREFIXES) or path.endswith(FORBIDDEN_TRACKED_SUFFIXES):
            offenders.append(path)
    return offenders


def tracked_forbidden_private_files(tracked_files: Sequence[str]) -> list[str]:
    offenders = []
    for path in tracked_files:
        if path in FORBIDDEN_PRIVATE_TRACKED_EXACT or path.startswith(
            FORBIDDEN_PRIVATE_TRACKED_PREFIXES
        ):
            offenders.append(path)
    return offenders


def missing_required_files(root: Path, tracked_files: Sequence[str]) -> list[str]:
    tracked = set(tracked_files)
    missing = []
    for rel_path in sorted(REQUIRED_TRACKED_FILES):
        if rel_path not in tracked or not (root / rel_path).is_file():
            missing.append(rel_path)
    return missing


def env_example_keys(root: Path) -> set[str]:
    return set(env_example_values(root))


def env_example_values(root: Path) -> dict[str, str]:
    env_path = root / ".env.example"
    if not env_path.exists():
        return {}
    values = {}
    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip()
    return values


def env_template_nonempty_private_values(root: Path) -> dict[str, str]:
    values = env_example_values(root)
    private_keys = PRIVATE_ENV_TEMPLATE_KEYS | external_resource_env_keys(root)
    return {key: value for key, value in values.items() if key in private_keys and value}


def _load_external_resources_manifest(root: Path) -> object:
    manifest_path = root / EXTERNAL_RESOURCES_MANIFEST
    if not manifest_path.exists():
        return None
    with manifest_path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def external_resource_env_keys(root: Path) -> set[str]:
    manifest = _load_external_resources_manifest(root)
    if not isinstance(manifest, dict):
        return set()
    resources = manifest.get("resources")
    if not isinstance(resources, list):
        return set()
    keys = set()
    for item in resources:
        if not isinstance(item, dict):
            continue
        env_key = item.get("env_key")
        if isinstance(env_key, str) and env_key.strip():
            keys.add(env_key.strip())
    return keys


def external_resource_manifest_errors(root: Path) -> list[str]:
    manifest_path = root / EXTERNAL_RESOURCES_MANIFEST
    if not manifest_path.exists():
        return [f"{EXTERNAL_RESOURCES_MANIFEST} is missing"]

    try:
        manifest = _load_external_resources_manifest(root)
    except json.JSONDecodeError as exc:
        return [f"{EXTERNAL_RESOURCES_MANIFEST} is invalid JSON: {exc}"]

    if not isinstance(manifest, dict):
        return [f"{EXTERNAL_RESOURCES_MANIFEST} must be a JSON object"]
    errors: list[str] = []
    if int(manifest.get("schema_version") or 0) != 1:
        errors.append("schema_version must be 1")

    resources = manifest.get("resources")
    if not isinstance(resources, list) or not resources:
        errors.append("resources must be a non-empty list")
        return errors

    seen_keys = set()
    for index, item in enumerate(resources):
        if not isinstance(item, dict):
            errors.append(f"resources[{index}] must be an object")
            continue
        env_key = str(item.get("env_key") or "").strip()
        if not env_key:
            errors.append(f"resources[{index}].env_key is required")
        elif env_key in seen_keys:
            errors.append(f"resources[{index}].env_key duplicates {env_key}")
        else:
            seen_keys.add(env_key)
        for field in ("category", "repo_policy"):
            if not str(item.get(field) or "").strip():
                errors.append(f"resources[{index}].{field} is required")
        if "value" in item or "default_value" in item:
            errors.append(f"resources[{index}] must not store value or default_value")
        if "read_only" in item and not isinstance(item["read_only"], bool):
            errors.append(f"resources[{index}].read_only must be boolean")
        if item.get("tracked_in_repo") is True:
            errors.append(f"resources[{index}].tracked_in_repo must not be true for external resources")
    return errors


def run_open_source_audit(root: Path, tracked_files: Sequence[str]) -> dict[str, object]:
    issues = audit_files(root, tracked_files)
    return build_open_source_report(issues)


def run_functional_surface_parity(root: Path, source_root: Path) -> dict[str, object]:
    report = build_functional_parity_report(source_root, root)
    return asdict(report)


def evaluate_requirements(
    root: Path,
    source_root: Path,
    tracked_files: Sequence[str],
    remotes: Sequence[str],
    open_source_report: dict[str, object],
    functional_parity_report: dict[str, object],
    external_manifest_errors: Sequence[str] | None = None,
) -> list[RequirementCheck]:
    root = root.resolve()
    source_root = source_root.resolve()
    checks: list[RequirementCheck] = []

    checks.append(
        RequirementCheck(
            name="source_root_exists",
            status="pass" if source_root.is_dir() else "fail",
            message="Source project root must exist and remain separate from PlantAI-rag.",
            evidence=str(source_root),
        )
    )
    checks.append(
        RequirementCheck(
            name="working_copy_is_separate",
            status="pass" if root != source_root else "fail",
            message="PlantAI-rag must not be the source project path.",
            evidence={"root": str(root), "source_root": str(source_root)},
        )
    )

    missing_files = missing_required_files(root, tracked_files)
    checks.append(
        RequirementCheck(
            name="required_runtime_files_tracked",
            status="pass" if not missing_files else "fail",
            message="Current runtime, validation, audit, and tests must be tracked.",
            evidence=missing_files,
        )
    )

    missing_env = sorted(REQUIRED_ENV_KEYS - env_example_keys(root))
    checks.append(
        RequirementCheck(
            name="env_template_has_required_keys",
            status="pass" if not missing_env else "fail",
            message=".env.example must expose required local configuration keys without values.",
            evidence=missing_env,
        )
    )

    manifest_errors = list(
        external_manifest_errors
        if external_manifest_errors is not None
        else external_resource_manifest_errors(root)
    )
    checks.append(
        RequirementCheck(
            name="external_resources_manifest_is_valid",
            status="pass" if not manifest_errors else "fail",
            message="External resource manifest must list env-managed resources without real paths or defaults.",
            evidence=manifest_errors,
        )
    )

    missing_external_env = sorted(external_resource_env_keys(root) - env_example_keys(root))
    checks.append(
        RequirementCheck(
            name="external_resource_env_keys_in_template",
            status="pass" if not missing_external_env else "fail",
            message="Every external resource env key must be documented in .env.example.",
            evidence=missing_external_env,
        )
    )

    nonempty_private_env = env_template_nonempty_private_values(root)
    checks.append(
        RequirementCheck(
            name="env_template_private_values_are_empty",
            status="pass" if not nonempty_private_env else "fail",
            message=".env.example must not contain private paths, credentials, or external resource defaults.",
            evidence=nonempty_private_env,
        )
    )

    forbidden_tracked = tracked_forbidden_data_files(tracked_files)
    checks.append(
        RequirementCheck(
            name="no_tracked_data_or_experiment_files",
            status="pass" if not forbidden_tracked else "fail",
            message="Tracked baseline must not include raw data, papers, model weights, or experiment outputs.",
            evidence=forbidden_tracked,
        )
    )

    forbidden_private = tracked_forbidden_private_files(tracked_files)
    checks.append(
        RequirementCheck(
            name="no_tracked_private_env_or_agent_state",
            status="pass" if not forbidden_private else "fail",
            message="Tracked baseline must not include private env files, agent state, or CodeGraph indexes.",
            evidence=forbidden_private,
        )
    )

    checks.append(
        RequirementCheck(
            name="local_git_has_no_remote",
            status="pass" if not remotes else "fail",
            message="Cleanup baseline must remain local-only until explicitly published.",
            evidence=list(remotes),
        )
    )

    open_source_summary = open_source_report.get("summary", {})
    blocking_runtime = int(open_source_summary.get("blocking_current_runtime", 0))  # type: ignore[union-attr]
    checks.append(
        RequirementCheck(
            name="open_source_current_runtime_has_no_blockers",
            status="pass" if blocking_runtime == 0 else "fail",
            message="Current runtime surface must have no open-source audit blockers.",
            evidence=open_source_summary,
        )
    )

    missing_functional = int(functional_parity_report.get("missing_count", 0))
    checks.append(
        RequirementCheck(
            name="functional_surface_has_no_missing_code",
            status="pass" if missing_functional == 0 else "fail",
            message="PlantAI-rag must retain source project code/config functional surface while excluding data and outputs.",
            evidence={
                "source_count": functional_parity_report.get("source_count"),
                "target_count": functional_parity_report.get("target_count"),
                "missing_count": missing_functional,
                "missing_entrypoint_count": functional_parity_report.get("missing_entrypoint_count"),
                "missing": list(functional_parity_report.get("missing", []))[:20],
                "missing_entrypoints": list(functional_parity_report.get("missing_entrypoints", []))[:20],
            },
        )
    )

    return checks


def build_report(root: Path, source_root: Path) -> dict[str, object]:
    tracked = git_tracked_files(root)
    remotes = git_remotes(root)
    open_source_report = run_open_source_audit(root, tracked)
    functional_parity_report = run_functional_surface_parity(root, source_root)
    external_manifest_errors = external_resource_manifest_errors(root)
    checks = evaluate_requirements(
        root,
        source_root,
        tracked,
        remotes,
        open_source_report,
        functional_parity_report,
        external_manifest_errors,
    )
    failures = [check for check in checks if check.status == "fail"]
    return {
        "summary": {
            "checks": len(checks),
            "failures": len(failures),
            "status": "pass" if not failures else "fail",
        },
        "checks": [asdict(check) for check in checks],
        "open_source_audit_summary": open_source_report.get("summary", {}),
        "functional_surface_parity_summary": {
            "source_count": functional_parity_report.get("source_count"),
            "target_count": functional_parity_report.get("target_count"),
            "missing_count": functional_parity_report.get("missing_count"),
            "missing_entrypoint_count": functional_parity_report.get("missing_entrypoint_count"),
        },
    }


def print_text_report(report: dict[str, object]) -> None:
    print("PlantAI-rag migration requirements audit")
    print(json.dumps(report["summary"], ensure_ascii=False, indent=2))
    for item in report["checks"]:  # type: ignore[index]
        check = RequirementCheck(**item)
        marker = "PASS" if check.status == "pass" else "FAIL"
        print(f"- {marker} {check.name}: {check.message}")
        if check.status == "fail":
            print(f"  evidence: {json.dumps(check.evidence, ensure_ascii=False)}")


def main(argv: Iterable[str] | None = None) -> int:
    args = parse_args(argv)
    source_root = resolve_source_root(args.source_root)
    report = build_report(args.root.resolve(), source_root.resolve())
    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2))
    else:
        print_text_report(report)
    failed = int(report["summary"]["failures"])  # type: ignore[index]
    if failed and not args.no_fail:
        return 2
    return 0


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