143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Evaluate AIOS V2R1C4 dormant release evidence.
|
|
|
|
This command performs no Git, network, deployment, tag, Registry or Baseline
|
|
write. It evaluates explicit evidence supplied by a governed execution agent
|
|
and writes a deterministic JSON decision report.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any, Mapping
|
|
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
PACKAGE_ROOT = REPOSITORY_ROOT / "services" / "aios-autonomous-release-candidate"
|
|
if str(PACKAGE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PACKAGE_ROOT))
|
|
|
|
from aios_autonomous_release_candidate.acceptance import ( # noqa: E402
|
|
AcceptancePhase,
|
|
AcceptanceStatus,
|
|
DormantAcceptanceGate,
|
|
DormantAcceptanceInput,
|
|
FinalDormantAcceptanceGate,
|
|
FinalDormantAcceptanceInput,
|
|
)
|
|
from aios_autonomous_release_candidate.models import sha256_json # noqa: E402
|
|
|
|
|
|
REPORT_SCHEMA = "aios.dormant-release.acceptance/v1"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RunnerDecision:
|
|
phase: str
|
|
status: str
|
|
report: Mapping[str, Any]
|
|
exit_code: int
|
|
|
|
|
|
def read_json_object(path: Path) -> Mapping[str, Any]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise ValueError(f"input file does not exist: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"input is not valid JSON: {exc.msg}") from exc
|
|
if not isinstance(value, dict):
|
|
raise ValueError("input must contain one JSON object")
|
|
return value
|
|
|
|
|
|
def atomic_write_json(path: Path, payload: Mapping[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(payload, sort_keys=True, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.chmod(0o600)
|
|
temporary.replace(path)
|
|
|
|
|
|
def evaluate(phase: AcceptancePhase, evidence: Mapping[str, Any]) -> RunnerDecision:
|
|
if phase is AcceptancePhase.CANDIDATE:
|
|
parsed = DormantAcceptanceInput.from_mapping(evidence)
|
|
result = DormantAcceptanceGate().evaluate(parsed)
|
|
else:
|
|
parsed = FinalDormantAcceptanceInput.from_mapping(evidence)
|
|
result = FinalDormantAcceptanceGate().evaluate(parsed)
|
|
result_payload = dict(result.payload())
|
|
report = {
|
|
"schema": REPORT_SCHEMA,
|
|
"phase": result.phase.value,
|
|
"status": result.status.value,
|
|
"decision": result_payload,
|
|
"input_sha256": sha256_json(evidence),
|
|
"production_action_authorized": False,
|
|
}
|
|
return RunnerDecision(
|
|
phase=result.phase.value,
|
|
status=result.status.value,
|
|
report=report,
|
|
exit_code=0 if result.status is AcceptanceStatus.READY else 3,
|
|
)
|
|
|
|
|
|
def invalid_report(phase: str, message: str) -> RunnerDecision:
|
|
report = {
|
|
"schema": REPORT_SCHEMA,
|
|
"phase": phase,
|
|
"status": AcceptanceStatus.BLOCKED.value,
|
|
"decision": {
|
|
"blocking_code": "INVALID_ACCEPTANCE_INPUT",
|
|
"blocking_message": message,
|
|
},
|
|
"production_action_authorized": False,
|
|
}
|
|
return RunnerDecision(
|
|
phase=phase,
|
|
status=AcceptanceStatus.BLOCKED.value,
|
|
report=report,
|
|
exit_code=2,
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="run_acceptance.py")
|
|
parser.add_argument(
|
|
"--phase",
|
|
required=True,
|
|
choices=[phase.value for phase in AcceptancePhase],
|
|
)
|
|
parser.add_argument("--input", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
phase = AcceptancePhase(args.phase)
|
|
evidence = read_json_object(args.input)
|
|
decision = evaluate(phase, evidence)
|
|
except (OSError, TypeError, ValueError) as exc:
|
|
decision = invalid_report(args.phase, str(exc))
|
|
|
|
envelope = dict(decision.report)
|
|
envelope["generated_at"] = datetime.now(timezone.utc).isoformat()
|
|
atomic_write_json(args.output, envelope)
|
|
print(json.dumps(envelope, sort_keys=True))
|
|
return decision.exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|