#!/usr/bin/env python3 """Generate the deterministic K8-REL-07 remediation intake. The intake turns the frozen K8 release blockers into independently traceable work items. It is an orchestration artifact: a verified intake does not resolve a blocker or make the package eligible for release. """ from __future__ import annotations import argparse import hashlib import json from pathlib import Path from typing import Any CONTEXT = Path("context-lock.json") CLOSEOUT = Path("coverage/k8-milestone-closeout.report.json") AUDIT = Path("coverage/post-release-audit.report.json") GATE = Path("coverage/release-candidate-gate.report.json") CLOSURE = Path("coverage/release-closure.report.json") INDEX = Path("coverage/release-evidence-index.json") OUTPUT = Path("coverage/release-remediation-intake.report.json") BLOCKER_TASKS = { "manifest_seed_unassigned": { "task_id": "K8-REL-08", "title": "manifest ownership remediation", }, "accepted_entries_missing": { "task_id": "K8-REL-09", "title": "release acceptance evidence remediation", }, "official_oracle_missing": { "task_id": "K8-REL-10", "title": "official oracle input remediation", }, "configuration_resources_missing": { "task_id": "K8-REL-11", "title": "configuration resource remediation", }, "browser_matrix_unverified": { "task_id": "K8-REL-12", "title": "browser matrix verification", }, "license_closure_incomplete": { "task_id": "K8-REL-13", "title": "implementation license closure", }, } 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 declared_report_hash(path: Path, report: dict[str, Any], error: str | None) -> dict[str, Any]: declared = report.get("report_sha256") actual = digest({key: value for key, value in report.items() if key != "report_sha256"}) if report else None status = "verified" if error is None and declared and declared == actual else "blocked" return { "path": path.as_posix(), "declared": declared, "actual": actual, "status": status, **({"error": error} if error else {}), } def compare_inputs(root: Path, closeout: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]: rows: list[dict[str, Any]] = [] all_match = isinstance(closeout.get("inputs"), list) for expected in closeout.get("inputs", []): path_text = expected.get("path") if isinstance(expected, dict) else None if not isinstance(path_text, str): rows.append({"path": None, "status": "blocked", "reason": "closeout input has no path"}) all_match = False continue path = Path(path_text) full = root / path actual_sha = file_digest(full) actual_bytes = full.stat().st_size if full.is_file() else None matches = ( actual_sha is not None and actual_sha == expected.get("sha256") and actual_bytes == expected.get("bytes") ) rows.append({ "path": path.as_posix(), "role": expected.get("role"), "expected_sha256": expected.get("sha256"), "actual_sha256": actual_sha, "expected_bytes": expected.get("bytes"), "actual_bytes": actual_bytes, "status": "verified" if matches else "blocked", }) all_match = all_match and matches return rows, all_match def context_check(lock: dict[str, Any], reports: dict[Path, dict[str, Any]]) -> dict[str, Any]: context_id = lock.get("context_id") observations = [] all_match = bool(context_id) for path, report in reports.items(): observed = report.get("context_id") matches = bool(observed) and observed == context_id observations.append({ "path": path.as_posix(), "context_id": observed, "status": "verified" if matches else "blocked", }) all_match = all_match and matches return { "context_id": context_id, "reports": observations, "status": "verified" if all_match else "blocked", } def blocker_check(root: Path, audit: dict[str, Any], gate: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]: audit_rows = audit.get("blockers") if isinstance(audit.get("blockers"), list) else [] gate_ids = {row.get("id") for row in gate.get("blockers", []) if isinstance(row, dict)} rows: list[dict[str, Any]] = [] all_valid = len(audit_rows) == len(BLOCKER_TASKS) and gate_ids == set(BLOCKER_TASKS) for sequence, row in enumerate(audit_rows, start=1): blocker_id = row.get("id") task = BLOCKER_TASKS.get(blocker_id, {}) evidence = row.get("evidence") if isinstance(row.get("evidence"), list) else [] evidence_paths = [item.get("path") for item in evidence if isinstance(item, dict)] evidence_present = bool(evidence_paths) and all( isinstance(item, dict) and item.get("status") == "present" and isinstance(item.get("path"), str) and (root / Path(item["path"])).is_file() for item in evidence ) valid = ( blocker_id in BLOCKER_TASKS and row.get("status") == "blocked" and row.get("owner") and row.get("review_owner") and row.get("remediation") and evidence_present and task.get("task_id") ) work_item = { "sequence": sequence, "id": blocker_id, "task_id": task.get("task_id"), "title": task.get("title"), "count": row.get("count"), "owner": row.get("owner"), "review_owner": row.get("review_owner"), "reason": row.get("reason"), "remediation": row.get("remediation"), "review_trigger": row.get("review_trigger"), "evidence": evidence, "status": "blocked", "release_effect": "STOP_RELEASE", "intake_status": "verified" if valid else "blocked", } rows.append(work_item) all_valid = all_valid and valid identity = {row.get("id") for row in rows} == set(BLOCKER_TASKS) checks = { "status": "verified" if all_valid and identity else "blocked", "expected_count": len(BLOCKER_TASKS), "observed_count": len(rows), "expected_ids": sorted(BLOCKER_TASKS), "observed_ids": sorted(str(row.get("id")) for row in rows), "identity_preserved": identity, "all_items_have_owner_and_evidence": all_valid, } return rows, checks def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]: root = root.resolve() paths = (CONTEXT, CLOSEOUT, AUDIT, GATE, CLOSURE, INDEX) documents: dict[Path, dict[str, Any]] = {} errors: dict[str, str] = {} for path in paths: documents[path], error = read_json(root, path) if error: errors[path.as_posix()] = error lock = documents[CONTEXT] closeout = documents[CLOSEOUT] audit = documents[AUDIT] gate = documents[GATE] reports_for_context = {path: documents[path] for path in (CLOSEOUT, AUDIT, GATE, CLOSURE, INDEX)} input_rows, inputs_match = compare_inputs(root, closeout) context = context_check(lock, reports_for_context) blocker_rows, blockers = blocker_check(root, audit, gate) report_hashes = [ declared_report_hash(path, documents[path], errors.get(path.as_posix())) for path in (CLOSEOUT, AUDIT, GATE, CLOSURE, INDEX) ] report_hashes_ok = all(row["status"] == "verified" for row in report_hashes) stop_decision = ( closeout.get("release_action") == "STOP_RELEASE" and closeout.get("release_decision") == "blocked_not_release_candidate" and closeout.get("candidate_eligible") is False and closeout.get("blocked_count") == 6 and gate.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False and gate.get("blocked_count") == 6 and audit.get("release_action") == "STOP_RELEASE" and audit.get("candidate_eligible") is False and audit.get("blocked_count") == 6 ) no_reclassification = all( row.get("status") == "blocked" and row.get("release_effect") == "STOP_RELEASE" for row in blocker_rows ) and len(blocker_rows) == 6 checks = { "input_hashes": {"status": "verified" if inputs_match else "blocked", "comparisons": input_rows}, "source_report_self_hashes": {"status": "verified" if report_hashes_ok else "blocked", "reports": report_hashes}, "context_consistency": context, "blocker_identity": blockers, "stop_decision_preserved": {"status": "verified" if stop_decision else "blocked", "release_action": "STOP_RELEASE" if stop_decision else None}, "no_reclassification": {"status": "verified" if no_reclassification else "blocked"}, } checks_ok = not errors and all(value.get("status") == "verified" for value in checks.values()) report: dict[str, Any] = { "schema": "release-remediation-intake.v1", "context_id": lock.get("context_id"), "intake_id": "K8-REL-07", "generated_at": lock.get("generated_at"), "status": "complete" if checks_ok else "blocked", "intake_decision": "remediation_required", "release_decision": "blocked_not_release_candidate", "release_action": "STOP_RELEASE", "candidate_eligible": False, "source_baseline": { "linuxcnc_version": lock.get("linuxcnc_version"), "linuxcnc_source_sha": lock.get("linuxcnc_source_sha"), "context_lock": artifact(root, CONTEXT, "context_lock"), }, "inputs": [ artifact(root, CLOSEOUT, "k8_milestone_closeout"), artifact(root, AUDIT, "post_release_audit"), artifact(root, GATE, "release_candidate_gate"), artifact(root, CLOSURE, "release_closure"), artifact(root, INDEX, "release_evidence_index"), artifact(root, CONTEXT, "context_lock"), ], "checks": checks, "work_items": blocker_rows, "blocked_count": len(blocker_rows), "next_tasks": [ {"task_id": row["task_id"], "title": row["title"], "depends_on": ["K8-REL-07"], "status": "pending"} for row in blocker_rows ], "rebuild": { "commands": [ "python3 scripts/generate_release_closure.py", "python3 scripts/generate_post_release_audit.py", "python3 scripts/generate_k8_milestone_closeout.py", "python3 scripts/generate_release_remediation_intake.py", "python3 -m unittest discover -s scripts -p 'test_*.py' -v", ], "deterministic_inputs": [path.as_posix() for path in paths], }, "failure_policy": { "on_missing_input": "STOP_RELEASE", "on_hash_mismatch": "STOP_RELEASE", "on_context_drift": "STOP_RELEASE", "forbidden": ["delete_samples", "lower_checks", "reclassify_as_REAL-EXCLUDED", "promote_blocked_to_accepted"], }, "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("release_remediation_intake=" + report["status"] + " action=" + report["release_action"] + " blockers=" + str(report["blocked_count"])) return 0 if __name__ == "__main__": raise SystemExit(main())