#!/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())