linuxcncWebWASM/scripts/generate_owner_review_gate_refresh.py

#!/usr/bin/env python3
"""Record the K8-REL-15 owner-review gate refresh and fail closed if inputs are absent."""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
from typing import Any


CONTEXT = Path("context-lock.json")
ROUNDUP = Path("coverage/remediation-evidence-roundup.report.json")
GATE = Path("coverage/release-candidate-gate.report.json")
AUDIT = Path("coverage/post-release-audit.report.json")
OUTPUT = Path("coverage/owner-review-gate-refresh.report.json")

DECISIONS = (
    ("manifest_seed_unassigned", "approve manifest ownership mapping and any new context"),
    ("accepted_entries_missing", "supply acceptance evidence with a ready official oracle"),
    ("official_oracle_missing", "add or approve locked kinematics and recovery oracle inputs"),
    ("configuration_resources_missing", "review and resolve configuration resource candidates"),
    ("browser_matrix_unverified", "approve minimums and execute the three-browser matrix"),
    ("license_closure_incomplete", "approve SPDX/NOTICE policy for Web/WASM implementation"),
)


def canonical(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def digest(value: Any) -> str:
    return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()


def file_digest(path: Path) -> str | None:
    if not path.is_file():
        return None
    handle = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            handle.update(block)
    return handle.hexdigest()


def read_json(root: Path, path: Path) -> tuple[dict[str, Any], str | None]:
    try:
        value = json.loads((root / path).read_text(encoding="utf-8"))
        if not isinstance(value, dict):
            raise ValueError(f"{path} must contain an object")
        return value, None
    except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
        return {}, str(exc)


def artifact(root: Path, path: Path, role: str) -> dict[str, Any]:
    full = root / path
    sha = file_digest(full)
    if sha is None:
        return {"path": path.as_posix(), "role": role, "status": "missing"}
    return {"path": path.as_posix(), "role": role, "status": "present", "bytes": full.stat().st_size, "sha256": sha}


def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]:
    root = root.resolve()
    lock, lock_error = read_json(root, CONTEXT)
    roundup, roundup_error = read_json(root, ROUNDUP)
    gate, gate_error = read_json(root, GATE)
    audit, audit_error = read_json(root, AUDIT)
    errors = {path.as_posix(): error for path, error in ((CONTEXT, lock_error), (ROUNDUP, roundup_error), (GATE, gate_error), (AUDIT, audit_error)) if error}
    blocker_ids = {row.get("id") for row in audit.get("blockers", []) if isinstance(row, dict)}
    decisions = [
        {"blocker": blocker, "required_action": action, "owner_input": "missing", "status": "blocked"}
        for blocker, action in DECISIONS
    ]
    checks = {
        "roundup_complete": {"status": "verified" if roundup.get("status") == "complete" and roundup.get("roundup_status") == "blocked_pending_owner_actions" else "blocked"},
        "six_decisions_located": {"status": "verified" if len(decisions) == 6 and blocker_ids == {row["blocker"] for row in decisions} else "blocked", "count": len(decisions)},
        "owner_inputs_present": {"status": "blocked", "present": 0, "required": 6, "reason": "no owner approval or new locked input is present in the workspace"},
        "gate_refresh_allowed": {"status": "blocked", "reason": "refresh is forbidden until owner inputs and any new context are reviewed"},
        "release_stop_preserved": {"status": "verified" if gate.get("release_action") == "STOP_RELEASE" and audit.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False and audit.get("candidate_eligible") is False else "blocked", "release_action": "STOP_RELEASE" if gate.get("release_action") == "STOP_RELEASE" else None},
        "context_consistency": {"status": "verified" if lock.get("context_id") and roundup.get("context_id") == lock.get("context_id") else "blocked", "context_id": lock.get("context_id")},
    }
    report: dict[str, Any] = {
        "schema": "owner-review-gate-refresh.v1",
        "context_id": lock.get("context_id"),
        "task_id": "K8-REL-15",
        "status": "blocked",
        "review_status": "owner_input_required",
        "release_decision": "blocked_not_release_candidate",
        "release_action": "STOP_RELEASE",
        "candidate_eligible": False,
        "inputs": [artifact(root, CONTEXT, "context_lock"), artifact(root, ROUNDUP, "remediation_roundup"), artifact(root, GATE, "release_candidate_gate"), artifact(root, AUDIT, "post_release_audit")],
        "checks": checks,
        "decisions": decisions,
        "next_transition": "collect owner decisions, create a new context when locked inputs change, then regenerate closure/audit/gate",
        "errors": errors,
        "report_sha256": None,
    }
    report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"})
    destination = output if output.is_absolute() else root / output
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
    return report


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument("--output", type=Path, default=OUTPUT)
    args = parser.parse_args()
    report = generate(args.repo_root, args.output)
    print("owner_review_gate_refresh=" + report["status"] + " action=" + report["release_action"] + " decisions=" + str(len(report["decisions"])))
    return 0


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