#!/usr/bin/env python3 """Generate deterministic acceptance-readiness evidence for K8-REL-09.""" from __future__ import annotations import argparse import hashlib import json from collections import Counter from pathlib import Path from typing import Any CONTEXT = Path("context-lock.json") MANIFEST = Path("coverage/manifest.jsonl") INDEX = Path("coverage/release-evidence-index.json") GATE = Path("coverage/release-candidate-gate.report.json") INTAKE = Path("coverage/release-remediation-intake.report.json") OWNERSHIP = Path("coverage/manifest-ownership-remediation.report.json") OUTPUT = Path("coverage/release-acceptance-remediation.report.json") 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 read_jsonl(root: Path, path: Path) -> tuple[list[dict[str, Any]], str | None]: try: rows = [] with (root / path).open(encoding="utf-8") as stream: for line_number, line in enumerate(stream, 1): if not line.strip(): continue value = json.loads(line) if not isinstance(value, dict): raise ValueError(f"{path}:{line_number} must contain an object") rows.append(value) return rows, 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) index, index_error = read_json(root, INDEX) gate, gate_error = read_json(root, GATE) intake, intake_error = read_json(root, INTAKE) ownership, ownership_error = read_json(root, OWNERSHIP) manifest, manifest_error = read_jsonl(root, MANIFEST) errors = { path.as_posix(): error for path, error in ( (CONTEXT, lock_error), (INDEX, index_error), (GATE, gate_error), (INTAKE, intake_error), (OWNERSHIP, ownership_error), (MANIFEST, manifest_error), ) if error } affected = index.get("affected_manifest_entries", []) if isinstance(index.get("affected_manifest_entries"), list) else [] manifest_accepted = [ row for row in manifest if row.get("sim_status") == "accepted" or row.get("web_status") == "accepted" ] readiness_reasons: Counter[str] = Counter() ready_rows = [] blocked_rows = [] for row in affected: reasons = [] if not row.get("owner"): reasons.append("owner_missing") if not row.get("inputs"): reasons.append("inputs_missing") if not row.get("outputs"): reasons.append("outputs_missing") if not row.get("test_id"): reasons.append("test_missing") if row.get("oracle_status") != "ready": reasons.append("oracle_blocked") if row.get("context_id") != lock.get("context_id"): reasons.append("context_missing_or_mismatch") if reasons: blocked_rows.append({"id": row.get("id"), "reasons": reasons}) readiness_reasons.update(reasons) else: ready_rows.append(row.get("id")) unassigned = sum(1 for row in manifest if row.get("mapping_status") == "needs_owner") ownership_count = ownership.get("counts", {}).get("mapped") coverage_ok = ( len(manifest) == 8053 and len(affected) == 1804 and unassigned == 6249 and ownership_count == 6249 and len(ready_rows) == 0 and len(blocked_rows) == len(affected) ) stop_ok = ( gate.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False and gate.get("blocked_count") == 6 and intake.get("release_action") == "STOP_RELEASE" and intake.get("candidate_eligible") is False ) no_promotion = len(manifest_accepted) == 0 and all(row.get("oracle_status") == "blocked" for row in affected) checks = { "input_shape": {"status": "verified" if coverage_ok else "blocked", "manifest_total": len(manifest), "affected": len(affected)}, "acceptance_readiness": {"status": "verified" if len(ready_rows) == 0 else "blocked", "eligible_count": len(ready_rows), "blocked_count": len(blocked_rows), "reasons": dict(sorted(readiness_reasons.items()))}, "no_promotion": {"status": "verified" if no_promotion else "blocked", "manifest_accepted": len(manifest_accepted), "affected_oracle_blocked": all(row.get("oracle_status") == "blocked" for row in affected)}, "ownership_dependency": {"status": "verified" if ownership_count == 6249 and ownership.get("remediation_status") == "blocked_pending_manifest_review" else "blocked", "mapped": ownership_count}, "context_consistency": {"status": "verified" if lock.get("context_id") and index.get("context_id") == lock.get("context_id") else "blocked", "context_id": lock.get("context_id")}, "release_stop_preserved": {"status": "verified" if stop_ok else "blocked", "release_action": "STOP_RELEASE" if stop_ok else None}, } checks_ok = not errors and all(value["status"] == "verified" for value in checks.values()) report: dict[str, Any] = { "schema": "release-acceptance-remediation.v1", "context_id": lock.get("context_id"), "task_id": "K8-REL-09", "status": "complete" if checks_ok else "blocked", "remediation_status": "blocked_no_eligible_entries", "release_decision": "blocked_not_release_candidate", "release_action": "STOP_RELEASE", "candidate_eligible": False, "inputs": [ artifact(root, CONTEXT, "context_lock"), artifact(root, MANIFEST, "manifest_seed"), artifact(root, INDEX, "release_evidence_index"), artifact(root, OWNERSHIP, "manifest_ownership_remediation"), artifact(root, GATE, "release_candidate_gate"), artifact(root, INTAKE, "remediation_intake"), ], "counts": { "manifest_total": len(manifest), "manifest_accepted": len(manifest_accepted), "manifest_needs_owner": unassigned, "affected_entries": len(affected), "eligible_for_acceptance": len(ready_rows), "blocked_for_acceptance": len(blocked_rows), }, "blocked_reason_counts": dict(sorted(readiness_reasons.items())), "blocked_entry_ids": sorted(str(row["id"]) for row in blocked_rows), "checks": checks, "acceptance_policy": { "required": ["owner", "inputs", "outputs", "test_id", "ready oracle", "matching context_id", "implementation and difference evidence"], "forbidden": ["promote_blocked_to_accepted", "promote_derived_to_accepted", "delete_samples", "lower_checks", "reclassify_as_REAL-EXCLUDED"], }, "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_acceptance_remediation=" + report["status"] + " release=" + report["release_action"] + " eligible=" + str(report["counts"]["eligible_for_acceptance"])) return 0 if __name__ == "__main__": raise SystemExit(main())