#!/usr/bin/env python3 """Generate deterministic K8-REL-11 configuration resource remediation.""" 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") RESOURCES = Path("coverage/config-resources.jsonl") RESOURCE_REPORT = Path("coverage/config-resources.report.json") REGRESSION = Path("coverage/configuration-regression.report.json") GATE = Path("coverage/release-candidate-gate.report.json") INTAKE = Path("coverage/release-remediation-intake.report.json") OUTPUT = Path("coverage/configuration-resource-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 basename_token(row: dict[str, Any]) -> str: raw = str(row.get("raw_value", "")).strip().strip('"').strip("'") token = raw.split()[0] if raw else str(row.get("path", "")).split("/")[-1] token = token.removeprefix("LIB:") return token.split("/")[-1] def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]: root = root.resolve() resource_rows, resource_error = read_jsonl(root, RESOURCES) resource_report, resource_report_error = read_json(root, RESOURCE_REPORT) regression, regression_error = read_json(root, REGRESSION) lock, lock_error = read_json(root, CONTEXT) gate, gate_error = read_json(root, GATE) intake, intake_error = read_json(root, INTAKE) errors = { path.as_posix(): error for path, error in ( (RESOURCES, resource_error), (RESOURCE_REPORT, resource_report_error), (REGRESSION, regression_error), (CONTEXT, lock_error), (GATE, gate_error), (INTAKE, intake_error), ) if error } file_index: dict[str, list[str]] = {} for path in sorted(root.rglob("*")): if path.is_file() and ".git" not in path.parts: file_index.setdefault(path.name, []).append(path.relative_to(root).as_posix()) missing = [row for row in resource_rows if row.get("parse_status") == "missing"] unresolved = [row for row in resource_rows if row.get("parse_status") in {"host_or_variable_path", "outside_repo", "unresolved"}] missing_rows = [] candidate_counts: Counter[str] = Counter() for row in sorted(missing, key=lambda item: str(item.get("resource_source_id"))): token = basename_token(row) candidates = file_index.get(token, []) classification = "candidate_basename_match" if candidates else "no_candidate" if any(marker in str(row.get("raw_value", "")) for marker in ("LIB:", "\\", " -", " --", "WORKINGFOLDER")): classification = "candidate_requires_reference_resolution" if candidates else "reference_requires_resolution" candidate_counts[classification] += 1 missing_rows.append({ "resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "path": row.get("path"), "raw_value": row.get("raw_value"), "edge_kind": row.get("edge_kind"), "resource_kind": row.get("resource_kind"), "parse_status": row.get("parse_status"), "basename_token": token, "candidate_paths": candidates[:8], "remediation_status": "blocked_pending_reference_review", }) unresolved_rows = [ { "resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "path": row.get("path"), "raw_value": row.get("raw_value"), "edge_kind": row.get("edge_kind"), "parse_status": row.get("parse_status"), "remediation_status": "blocked_unresolved_reference", } for row in sorted(unresolved, key=lambda item: str(item.get("resource_source_id"))) ] config_entries = regression.get("configuration_entries", []) if isinstance(regression.get("configuration_entries"), list) else [] unindexed_entries = [ {"config_id": row.get("config_id"), "path": row.get("path"), "startup_preflight": row.get("startup_preflight"), "resource_status": row.get("resource_status")} for row in config_entries if row.get("resource_status") == "unindexed" ] unregistered_entries = [ {"config_id": row.get("config_id"), "path": row.get("path"), "module": row.get("module"), "startup_preflight": row.get("startup_preflight")} for row in config_entries if row.get("startup_preflight") == "blocked_unregistered_kinematics" ] counts = regression.get("counts", {}) expected_missing = counts.get("resource_missing_rows") expected_unresolved = counts.get("resource_unresolved_rows") expected_unindexed = counts.get("resource_status", {}).get("unindexed") all_retained = len(missing_rows) == expected_missing and len(unresolved_rows) == expected_unresolved and len(unindexed_entries) == expected_unindexed 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 ) checks = { "resource_rows_retained": {"status": "verified" if all_retained else "blocked", "missing": len(missing_rows), "unresolved": len(unresolved_rows), "unindexed": len(unindexed_entries)}, "candidate_resolution_is_non_mutating": {"status": "verified" if sum(candidate_counts.values()) == len(missing_rows) and all(row["remediation_status"] != "resolved" for row in missing_rows) else "blocked", "classification": dict(sorted(candidate_counts.items()))}, "startup_blockers_retained": {"status": "verified" if len(unregistered_entries) == counts.get("startup_preflight", {}).get("blocked_unregistered_kinematics") else "blocked", "unregistered": len(unregistered_entries)}, "resource_graph_closure": {"status": "verified" if regression.get("closure", {}).get("resource_orphan_entrypoints") == [] and regression.get("closure", {}).get("graph_orphan_entry_ids") == [] else "blocked"}, "context_consistency": {"status": "verified" if lock.get("context_id") and regression.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": "configuration-resource-remediation.v1", "context_id": lock.get("context_id"), "task_id": "K8-REL-11", "status": "complete" if checks_ok else "blocked", "remediation_status": "blocked_resource_gaps", "release_decision": "blocked_not_release_candidate", "release_action": "STOP_RELEASE", "candidate_eligible": False, "inputs": [ artifact(root, CONTEXT, "context_lock"), artifact(root, RESOURCES, "config_resources"), artifact(root, RESOURCE_REPORT, "config_resources_report"), artifact(root, REGRESSION, "configuration_regression"), artifact(root, GATE, "release_candidate_gate"), artifact(root, INTAKE, "remediation_intake"), ], "counts": { "resource_rows": len(resource_rows), "missing_rows": len(missing_rows), "unresolved_rows": len(unresolved_rows), "unindexed_configuration_entries": len(unindexed_entries), "unregistered_startup_entries": len(unregistered_entries), "candidate_basename_matches": sum(value for key, value in candidate_counts.items() if "candidate" in key), "resolved_by_this_artifact": 0, }, "candidate_classification": dict(sorted(candidate_counts.items())), "missing_rows": missing_rows, "unresolved_rows": unresolved_rows, "unindexed_entries": sorted(unindexed_entries, key=lambda row: str(row.get("config_id"))), "unregistered_entries": sorted(unregistered_entries, key=lambda row: str(row.get("config_id"))), "checks": checks, "review_policy": { "required_to_unblock": ["resolve LIB/command/variable references against an approved search path", "restore or explicitly remove missing resource references with source review", "index every include/template entrypoint", "register or explicitly block each kinematics module"], "forbidden": ["drop_missing_rows", "replace_missing_with_same_basename_without_contract", "reclassify_as_REAL-EXCLUDED", "promote_blocked_startup_to_ready", "overwrite_context_lock_in_place"], }, "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("configuration_resource_remediation=" + report["status"] + " release=" + report["release_action"] + " missing=" + str(report["counts"]["missing_rows"])) return 0 if __name__ == "__main__": raise SystemExit(main())