llm/scripts/check_1c_extension_validation_release.py

#!/usr/bin/env python3
"""Aggregate final validation gates for a staged 1C extension."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from check_1c_extension_staging import check_staging
from check_1c_extension_validation_evidence import check_evidence


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8-sig"))


def write_json(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def issue(severity: str, code: str, message: str, *, path: Path | str | None = None, detail: dict[str, Any] | None = None) -> dict[str, Any]:
    result: dict[str, Any] = {"severity": severity, "code": code, "message": message}
    if path is not None:
        result["path"] = str(path)
    if detail:
        result["detail"] = detail
    return result


def collect_gate(name: str, data: dict[str, Any]) -> dict[str, Any]:
    return {
        "name": name,
        "schema": data.get("schema"),
        "passed": data.get("passed"),
        "status": data.get("status"),
        "counts": data.get("counts"),
    }


def check_release(plan_path: Path, evidence_root: Path | None = None) -> dict[str, Any]:
    findings: list[dict[str, Any]] = []
    if not plan_path.exists():
        findings.append(issue("error", "missing_validation_plan", "Validation plan is missing.", path=plan_path))
        return build_result(plan_path, None, {}, {}, findings)

    plan = load_json(plan_path)
    if plan.get("schema") != "onec_extension_validation_plan.v1":
        findings.append(issue("error", "invalid_validation_plan_schema", "Validation plan schema is not onec_extension_validation_plan.v1.", path=plan_path, detail={"schema": plan.get("schema")}))

    if plan.get("status") != "ready_for_disposable_validation":
        findings.append(issue("error", "validation_plan_not_ready", "Validation plan must be ready_for_disposable_validation.", path=plan_path, detail={"status": plan.get("status")}))

    staging_dir = Path(str(plan.get("staging_dir") or ""))
    staging_check = check_staging(staging_dir) if staging_dir else {"schema": "onec_extension_staging_check.v1", "passed": False, "counts": {"errors": 1}, "findings": []}
    if not staging_check.get("passed"):
        findings.append(issue("error", "staging_check_failed", "Staging check failed.", path=staging_dir, detail={"counts": staging_check.get("counts")}))

    evidence_check = check_evidence(plan_path, evidence_root)
    if not evidence_check.get("passed"):
        findings.append(issue("error", "validation_evidence_check_failed", "Validation evidence check failed.", path=evidence_check.get("evidence_root"), detail={"counts": evidence_check.get("counts")}))

    safety = plan.get("safety") if isinstance(plan.get("safety"), dict) else {}
    expected_safety = {
        "production_base_allowed": False,
        "sql_write_allowed": False,
        "source_extension_write_allowed": False,
        "disposable_base_required": True,
    }
    for key, expected in expected_safety.items():
        if safety.get(key) is not expected:
            findings.append(issue("error", "invalid_release_safety_flag", "Validation plan safety flag has an unexpected value.", path=plan_path, detail={"flag": key, "expected": expected, "actual": safety.get(key)}))

    return build_result(plan_path, plan, staging_check, evidence_check, findings)


def build_result(plan_path: Path, plan: dict[str, Any] | None, staging_check: dict[str, Any], evidence_check: dict[str, Any], findings: list[dict[str, Any]]) -> dict[str, Any]:
    errors = [row for row in findings if row.get("severity") == "error"]
    warnings = [row for row in findings if row.get("severity") == "warning"]
    return {
        "schema": "onec_extension_validation_release_check.v1",
        "created_at_utc": datetime.now(timezone.utc).isoformat(),
        "plan_path": str(plan_path),
        "staging_dir": (plan or {}).get("staging_dir"),
        "bundle_dir": (plan or {}).get("bundle_dir"),
        "preferred_extension": (plan or {}).get("preferred_extension"),
        "passed": not errors,
        "status": "validated_for_human_review" if not errors else "blocked",
        "safety": {
            "production_apply_allowed": False,
            "automatic_apply_allowed": False,
            "human_approval_required": True,
        },
        "gates": [
            {
                "name": "validation_plan",
                "schema": (plan or {}).get("schema"),
                "passed": (plan or {}).get("status") == "ready_for_disposable_validation",
                "status": (plan or {}).get("status"),
            },
            collect_gate("staging_check", staging_check),
            collect_gate("validation_evidence_check", evidence_check),
        ],
        "findings": findings,
        "counts": {
            "errors": len(errors),
            "warnings": len(warnings),
        },
        "next_actions": [
            "Review validation evidence and changed files manually.",
            "Do not apply to production automatically.",
            "If approved, perform production action through the approved human-controlled 1C release process.",
        ] if not errors else [
            "Fix failed gates before review.",
            "Do not package, apply, or release this extension from the current evidence.",
        ],
        "details": {
            "staging_check": staging_check,
            "validation_evidence_check": evidence_check,
        },
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Aggregate final validation gates for a staged 1C extension.")
    parser.add_argument("--plan", type=Path, required=True)
    parser.add_argument("--evidence-root", type=Path)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()

    result = check_release(args.plan, args.evidence_root)
    if args.output:
        write_json(args.output, result)
    print(json.dumps({"output": str(args.output) if args.output else None, "passed": result["passed"], "status": result["status"], "counts": result["counts"]}, ensure_ascii=False))
    return 0 if result["passed"] else 2


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