#!/usr/bin/env python3 import json from pathlib import Path from datetime import datetime, timezone ROOT = Path("/home/yeff/public_html/devon") CANON_EXEC = ROOT / "canon" / "execution" STATE_DIR = CANON_EXEC / "state" REPORT = STATE_DIR / "execution_contract_validation_report.json" MASTER = CANON_EXEC / "master_execution_contract.json" REQUIRED_DIRS = [ CANON_EXEC / "schemas", CANON_EXEC / "categories", CANON_EXEC / "items", CANON_EXEC / "scans", CANON_EXEC / "refresh", CANON_EXEC / "progress", CANON_EXEC / "bin", CANON_EXEC / "state", ] REQUIRED_SCHEMA_FILES = [ CANON_EXEC / "schemas" / "category_execution_contract.schema.json", CANON_EXEC / "schemas" / "item_contract.schema.json", CANON_EXEC / "schemas" / "scan_contract.schema.json", CANON_EXEC / "schemas" / "refresh_event_contract.schema.json", CANON_EXEC / "schemas" / "progress_rule_contract.schema.json", CANON_EXEC / "schemas" / "missing_declaration_contract.schema.json", ] REQUIRED_FOUNDATION_FILES = [ CANON_EXEC / "categories" / "panel_canonical_tree.contract.json", CANON_EXEC / "items" / "panel_canonical_tree_001_verify_master_binding.contract.json", CANON_EXEC / "items" / "panel_canonical_tree_002_verify_category_contract.contract.json", CANON_EXEC / "items" / "panel_canonical_tree_003_verify_legacy_panel_absence.contract.json", CANON_EXEC / "scans" / "panel_canonical_tree_category_scan.contract.json", CANON_EXEC / "refresh" / "panel_canonical_tree_refresh.contract.json", CANON_EXEC / "progress" / "panel_canonical_tree_category_readiness.contract.json", CANON_EXEC / "state" / "panel_canonical_tree.resolved.json", ] def utc_now(): return datetime.now(timezone.utc).isoformat() def read_json(path): return json.loads(path.read_text(encoding="utf-8")) def status_record(path): rec = { "path": str(path), "exists": path.exists(), "json_status": "MISSING", "contract_id": "", "schema_id": "", "status": "MISSING", "error": "" } if not path.exists(): return rec try: data = read_json(path) rec["json_status"] = "PASS" rec["contract_id"] = ( data.get("contract_id") or data.get("category_contract_id") or data.get("item_contract_id") or data.get("scan_contract_id") or data.get("refresh_event_contract_id") or data.get("progress_rule_contract_id") or data.get("missing_declaration_contract_id") or data.get("state_id") or "" ) rec["schema_id"] = data.get("$id", "") rec["status"] = data.get("status") or data.get("final_status") or "PASS" except Exception as exc: rec["json_status"] = "FAIL" rec["status"] = "FAIL" rec["error"] = str(exc) return rec def collect_json_files(): roots = [ CANON_EXEC / "schemas", CANON_EXEC / "categories", CANON_EXEC / "items", CANON_EXEC / "scans", CANON_EXEC / "refresh", CANON_EXEC / "progress", CANON_EXEC / "state", ] files = [] for root in roots: if root.exists(): files.extend(sorted(root.glob("*.json"))) if MASTER.exists(): files.insert(0, MASTER) return files def forbidden_path_check(files): forbidden = [ str(ROOT / "panel"), str(ROOT / "panel" / "data"), ] findings = [] for path in files: try: text = path.read_text(encoding="utf-8") except Exception as exc: findings.append({ "path": str(path), "status": "FAIL", "error": str(exc) }) continue found = [x for x in forbidden if x in text] if found: findings.append({ "path": str(path), "status": "FAIL", "found": found }) return findings def validate_master(): result = { "status": "MISSING", "phase_count": 0, "category_count": 0, "panel_canonical_tree_status": "MISSING", "panel_canonical_tree_path": "", "error": "" } if not MASTER.exists(): return result try: master = read_json(MASTER) phases = master.get("phases", []) result["phase_count"] = len(phases) result["category_count"] = sum(len(p.get("categories", [])) for p in phases) for phase in phases: if phase.get("phase_id") != "phase-02": continue for category in phase.get("categories", []): if category.get("category_id") == "panel_canonical_tree": result["panel_canonical_tree_status"] = category.get("category_contract_status", "MISSING") result["panel_canonical_tree_path"] = category.get("category_contract_path", "") if result["phase_count"] == 14 and result["category_count"] == 152 and result["panel_canonical_tree_status"] == "ACTIVE": result["status"] = "PASS" else: result["status"] = "FAIL" except Exception as exc: result["status"] = "FAIL" result["error"] = str(exc) return result def validate_resolved_state(): result = { "status": "MISSING", "active_category_count": 0, "pass_count": 0, "fail_count": 0, "missing_count": 0, "states": [], "error": "" } if not MASTER.exists(): result["status"] = "MISSING" result["error"] = "master execution contract missing" return result try: master = read_json(MASTER) active_categories = [] for phase in master.get("phases", []): for category in phase.get("categories", []): if category.get("category_contract_status") == "ACTIVE": active_categories.append({ "phase_id": phase.get("phase_id", ""), "category_id": category.get("category_id", ""), "category_title": category.get("category_title", "") }) result["active_category_count"] = len(active_categories) for row in active_categories: category_id = row["category_id"] state_path = CANON_EXEC / "state" / f"{category_id}.resolved.json" state_row = { "phase_id": row["phase_id"], "category_id": category_id, "category_title": row["category_title"], "path": str(state_path), "status": "MISSING", "final_status": "MISSING", "progress_percent": 0, "scan_status": "MISSING", "refresh_status": "MISSING", "error": "" } if not state_path.exists(): result["missing_count"] += 1 result["states"].append(state_row) continue try: data = read_json(state_path) state_row["final_status"] = data.get("final_status", "MISSING") state_row["progress_percent"] = data.get("progress_percent", 0) state_row["scan_status"] = data.get("scan_status", "MISSING") state_row["refresh_status"] = data.get("refresh_status", "MISSING") if ( state_row["final_status"] == "PASS" and state_row["progress_percent"] == 100 and state_row["scan_status"] == "PASS" and state_row["refresh_status"] == "PASS" ): state_row["status"] = "PASS" result["pass_count"] += 1 else: state_row["status"] = "FAIL" result["fail_count"] += 1 except Exception as exc: state_row["status"] = "FAIL" state_row["error"] = str(exc) result["fail_count"] += 1 result["states"].append(state_row) if result["active_category_count"] == 0: result["status"] = "MISSING" elif result["fail_count"] == 0 and result["missing_count"] == 0: result["status"] = "PASS" else: result["status"] = "FAIL" except Exception as exc: result["status"] = "FAIL" result["error"] = str(exc) return result def main(): STATE_DIR.mkdir(parents=True, exist_ok=True) json_files = collect_json_files() dir_results = [ { "path": str(path), "status": "PASS" if path.exists() and path.is_dir() else "MISSING" } for path in REQUIRED_DIRS ] required_schema_results = [status_record(path) for path in REQUIRED_SCHEMA_FILES] foundation_results = [status_record(path) for path in REQUIRED_FOUNDATION_FILES] all_json_results = [status_record(path) for path in json_files] forbidden_findings = forbidden_path_check(json_files) master_result = validate_master() resolved_state_result = validate_resolved_state() blocking_reasons = [] if any(x["status"] != "PASS" for x in dir_results): blocking_reasons.append("required_execution_directory_missing") if any(x["json_status"] != "PASS" for x in required_schema_results): blocking_reasons.append("required_schema_invalid_or_missing") if any(x["json_status"] != "PASS" for x in foundation_results): blocking_reasons.append("foundation_contract_invalid_or_missing") if any(x["json_status"] != "PASS" for x in all_json_results): blocking_reasons.append("one_or_more_execution_json_files_invalid") if forbidden_findings: blocking_reasons.append("forbidden_legacy_absolute_path_found") if master_result["status"] != "PASS": blocking_reasons.append("master_execution_contract_not_ready") if resolved_state_result["status"] != "PASS": blocking_reasons.append("one_or_more_active_category_states_not_pass") final_status = "PASS" if not blocking_reasons else "FAIL" report = { "report_id": "execution_contract_validation_report", "version": "v1", "generated_at": utc_now(), "validator": "contract_validator.py", "final_status": final_status, "blocking_reasons": blocking_reasons, "summary": { "json_file_count": len(json_files), "required_directory_count": len(REQUIRED_DIRS), "required_schema_count": len(REQUIRED_SCHEMA_FILES), "foundation_contract_count": len(REQUIRED_FOUNDATION_FILES), "forbidden_path_finding_count": len(forbidden_findings) }, "master_result": master_result, "resolved_state_result": resolved_state_result, "required_directories": dir_results, "required_schemas": required_schema_results, "foundation_contracts": foundation_results, "forbidden_path_findings": forbidden_findings, "all_json_results": all_json_results, "next_required_action": "create_next_category_execution_contract" if final_status == "PASS" else "fix_execution_contract_validation_failures" } REPORT.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print("CONTRACT_VALIDATOR_STATUS=PASS") print("REPORT=", REPORT) print("FINAL_STATUS=", final_status) print("JSON_FILE_COUNT=", len(json_files)) print("FORBIDDEN_PATH_FINDINGS=", len(forbidden_findings)) print("BLOCKING_REASONS=", ",".join(blocking_reasons)) print("NEXT_REQUIRED_ACTION=", report["next_required_action"]) if __name__ == "__main__": main()