ai-exchange/scripts/run_automation_execution_closure_gate.py

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import argparse
import json
import os


ROOT = Path(__file__).resolve().parents[1]
TASKS_PATH = Path("TASKS.md")
SESSION_PATH = Path("SESSION.md")
TASKLIST_PATH = Path("docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md")

REQUIRED_QUEUE_ITEMS = (
    "`P15` Alert Noise Policy Matrix",
    "`P16` Browser Screenshot Capture Adapter",
    "`P17` Visual Trigger ROI Router",
    "`P18` Visual Evidence Media Dispatch",
    "`P19` Multi-Channel Alert Routing",
    "`P20` Operator Digest And Review Queue",
    "`P21` Runtime Incident Journal",
    "`P22` Maintenance And Retention Rollup",
    "`P23` End-To-End Unattended Harness",
    "`P24` Remote Smoke And Ops Closure",
    "`PXX` Automation Execution Closure Gate",
)

REQUIRED_GUARDS = (
    "execution_allowed",
    "real_live_ready",
    "risk_governor",
    "trust_bucket",
    "operator_confirmation_protocol",
)


def run_closure_gate(root: Path = ROOT) -> dict:
    tasks = _read_text(root / TASKS_PATH)
    session = _read_text(root / SESSION_PATH)
    tasklist = _read_text(root / TASKLIST_PATH)
    checks = [
        _check_queue_closed(tasks),
        _check_session_pxx_state(session),
        _check_scope_freeze_policy(tasklist),
        _check_validation_evidence(session),
        _check_guardrails(session),
    ]
    blockers = [check["name"] for check in checks if check["status"] == "blocked"]
    return {
        "status": "blocked" if blockers else "passed",
        "mode": "automation_execution_closure_gate",
        "generated_ts": datetime.now(timezone.utc).isoformat(),
        "checks": checks,
        "passed_checks": [check["name"] for check in checks if check["status"] == "passed"],
        "blockers": blockers,
        "post_pxx_state": {
            "state": "automation_execution_layer_pxx_known_scope_closed",
            "current_phase": "Long-running Analysis Decision Monitor With Ops Rollup",
            "next_action": "continue_periodic_analysis_decision_monitoring_with_maturity_rollup",
            "new_phase_policy": "require_explicit_user_product_decision",
        },
    }


def _check_queue_closed(tasks: str) -> dict:
    missing = []
    for item in REQUIRED_QUEUE_ITEMS:
        if f"- [x] {item}" not in tasks:
            missing.append(item)
    if missing:
        return _blocked("p15_pxx_queue_closed", f"unchecked queue items: {missing}")
    return _passed("p15_pxx_queue_closed", "P15-P24 and PXX are checked in TASKS.md")


def _check_session_pxx_state(session: str) -> dict:
    required = (
        "state: automation_execution_layer_pxx_known_scope_closed",
        "current_phase: Long-running Analysis Decision Monitor With Ops Rollup",
        "next_execution_layer_phase: Long-running Analysis Decision Monitor With Ops Rollup",
    )
    missing = [item for item in required if item not in session]
    if missing:
        return _blocked("session_pxx_state", f"SESSION.md missing {missing}")
    return _passed("session_pxx_state", "SESSION.md records the PXX frozen state")


def _check_scope_freeze_policy(tasklist: str) -> dict:
    required = (
        "new_phase_policy: require_explicit_user_product_decision",
        "不再自动发明 P25/P26",
    )
    missing = [item for item in required if item not in tasklist]
    if missing:
        return _blocked("scope_freeze_policy", f"tasklist missing {missing}")
    return _passed("scope_freeze_policy", "PXX tasklist freezes automatic phase creation")


def _check_validation_evidence(session: str) -> dict:
    required = (
        "PXX final validation passed",
        "unit tests",
        "script static checks",
        "P24 remote ops smoke passed",
    )
    missing = [item for item in required if item not in session]
    if missing:
        return _blocked("validation_evidence", f"SESSION.md missing {missing}")
    return _passed("validation_evidence", "SESSION.md records final quality and remote smoke evidence")


def _check_guardrails(session: str) -> dict:
    missing = [guard for guard in REQUIRED_GUARDS if guard not in session]
    if missing:
        return _blocked("guardrails_unchanged", f"SESSION.md missing guards {missing}")
    return _passed("guardrails_unchanged", "guardrail names remain recorded as unchanged")


def _read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def _passed(name: str, detail: str) -> dict:
    return {"name": name, "status": "passed", "detail": detail}


def _blocked(name: str, detail: str) -> dict:
    return {"name": name, "status": "blocked", "detail": detail}


def _get_bool(env, key: str, default: bool) -> bool:
    value = env.get(key)
    if value is None or value == "":
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


def _parse_args(argv: Optional[list[str]]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run the PXX automation execution closure gate.")
    parser.add_argument("--root", default=str(ROOT))
    return parser.parse_args(argv)


def main(argv: Optional[list[str]] = None, env=None) -> int:
    env = os.environ if env is None else env
    args = _parse_args(argv)
    payload = run_closure_gate(root=Path(args.root))
    if _get_bool(env, "CLOSURE_GATE_OUTPUT_JSON", False):
        print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
    else:
        print(f"Closure gate status: {payload['status']}")
        print(f"Passed checks: {', '.join(payload['passed_checks']) or 'none'}")
        print(f"Blockers: {', '.join(payload['blockers']) or 'none'}")
    return 0 if payload["status"] == "passed" else 2


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