#!/usr/bin/env python3 import json import hashlib from pathlib import Path from datetime import datetime, timezone DEVON_ROOT = Path("/home/yeff/public_html/devon") EXEC_ROOT = DEVON_ROOT / "canon" / "execution" STATE_DIR = EXEC_ROOT / "state" MASTER_PATH = EXEC_ROOT / "master_execution_contract.json" VALIDATION_REPORT_PATH = STATE_DIR / "execution_contract_validation_report.json" GENERIC_SUMMARY_PATH = STATE_DIR / "generic_category_state_resolution_summary.json" EXPANSION_PLAN_PATH = STATE_DIR / "contract_expansion_plan.json" BULK_REPORT_PATH = STATE_DIR / "category_bulk_expansion_report.json" OUTPUT_PATH = STATE_DIR / "panel_bootstrap_index.json" FORBIDDEN_ABSOLUTE_PANEL_PATH = str(DEVON_ROOT / "panel") def utc_now(): return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def read_json(path, required=False): if not path.exists(): if required: raise FileNotFoundError(str(path)) return {} with path.open("r", encoding="utf-8") as fh: return json.load(fh) def sha256_file(path): if not path.exists(): return "" h = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def recursive_find_key(obj, names): if isinstance(names, str): names = {names} else: names = set(names) if isinstance(obj, dict): for key in names: if key in obj and obj[key] not in (None, ""): return obj[key] for value in obj.values(): found = recursive_find_key(value, names) if found not in (None, ""): return found if isinstance(obj, list): for item in obj: found = recursive_find_key(item, names) if found not in (None, ""): return found return None def normalize_contract_status(value): if not value: return "MISSING" value = str(value).upper() if value in {"PASS", "ACTIVE"}: return value if value in {"FAIL", "FAILED"}: return "FAIL" if value in {"MISSING", "DRAFT", "PENDING"}: return value return value def normalize_final_status(value): if not value: return "MISSING" value = str(value).upper() if value in {"PASS", "FAIL", "MISSING"}: return value if value == "ACTIVE": return "PASS" return "MISSING" def find_forbidden_path_findings(): findings = [] for path in EXEC_ROOT.rglob("*.json"): if path == OUTPUT_PATH: continue try: text = path.read_text(encoding="utf-8", errors="ignore") except Exception: continue if FORBIDDEN_ABSOLUTE_PANEL_PATH in text: findings.append(str(path)) return findings def count_json_files_excluding_output(): count = 0 for path in EXEC_ROOT.rglob("*.json"): if path == OUTPUT_PATH: continue count += 1 return count def build_resolved_state_map(): mapping = {} for path in STATE_DIR.glob("*.resolved.json"): data = read_json(path, required=False) category_id = ( recursive_find_key(data, ["category_id", "source_category_id"]) or path.name.replace(".resolved.json", "") ) if category_id: mapping[str(category_id)] = { "path": path, "data": data, } return mapping def safe_rel(path): if not path: return "" return str(path) def get_report_status(report): value = recursive_find_key(report, [ "final_status", "validator_status", "validation_status", "status", "result_status", ]) return normalize_final_status(value) def get_planner_status(report): value = recursive_find_key(report, [ "planner_status", "status", "result_status", "final_status", ]) return normalize_final_status(value) def get_expansion_status(report): value = recursive_find_key(report, [ "expansion_status", "status", "planner_expansion_status", ]) if not value: return "MISSING" return str(value).upper() def get_candidate_category_count(report): value = recursive_find_key(report, [ "candidate_category_count", "candidate_count", "remaining_candidate_count", ]) if isinstance(value, int): return value try: return int(value) except Exception: return 0 def get_category_contract_path(category): value = category.get("category_contract_path") or category.get("contract_path") if value: return value category_id = category.get("category_id") or category.get("source_category_id") if not category_id: return "" direct = EXEC_ROOT / "categories" / f"{category_id}.contract.json" if direct.exists(): return str(direct) matches = list((EXEC_ROOT / "categories").glob(f"*{category_id}*.contract.json")) if matches: return str(matches[0]) return str(direct) def get_resolved_status(category_id, resolved_map): row = resolved_map.get(str(category_id)) if not row: return "MISSING" data = row["data"] status = recursive_find_key(data, [ "final_status", "resolved_status", "status", "result_status", ]) return normalize_final_status(status) def get_resolved_progress(category_id, resolved_map, fallback_status): row = resolved_map.get(str(category_id)) if row: data = row["data"] value = recursive_find_key(data, [ "progress_percent", "progress_pct", "current_percent", ]) try: return int(value) except Exception: pass if fallback_status == "PASS": return 100 return 0 def get_blocking_reasons(category_id, resolved_map): row = resolved_map.get(str(category_id)) if not row: return ["resolved_state_missing"] data = row["data"] for key in ["blocking_reasons", "blocks", "missing_to_100", "failures"]: value = recursive_find_key(data, key) if isinstance(value, list): return value if isinstance(value, str) and value: return [value] status = get_resolved_status(category_id, resolved_map) if status == "PASS": return [] return ["resolved_state_not_pass"] def count_item_contracts_for_category(category_id): item_dir = EXEC_ROOT / "items" if not item_dir.exists(): return 0 count = 0 for path in item_dir.rglob("*.json"): matched = False if str(category_id) in path.name: matched = True else: data = read_json(path, required=False) source_category = recursive_find_key(data, ["category_id", "source_category_id"]) if str(source_category) == str(category_id): matched = True if matched: count += 1 return count def get_last_category_remote_verify_job(): jobs_path = STATE_DIR / "yeffai_bootstrap_jobs.json" if not jobs_path.exists(): return None try: data = json.loads(jobs_path.read_text(encoding="utf-8")) except Exception: return None jobs = data.get("jobs") if not isinstance(jobs, list): return None for job in reversed(jobs): if isinstance(job, dict) and job.get("action") == "category_remote_verify": return job return None def category_remote_verify_runtime_fields(): job = get_last_category_remote_verify_job() if not isinstance(job, dict): return { "yeffai_remote_verification_status": "MISSING", "remote_verification_status": "MISSING", "category_remote_verification_status": "MISSING", "yeffai_category_remote_status": "MISSING", "remote_execution_status": "MISSING", "operational_execution_status": "MISSING", "worker_status": "MISSING", "job_status": "MISSING", "last_job_status": "MISSING", "execution_job_status": "MISSING", "execution_status_remote": "MISSING", "category_remote_verify_job_id": "MISSING", "category_remote_verify_class": "MISSING", "category_remote_verify_categories_total": 0, "category_remote_verify_categories_missing": 0, "category_remote_verify_mutation": "NO", } result = job.get("result") if isinstance(job.get("result"), dict) else {} remote_status = str(result.get("remote_verification_state_status") or "MISSING") job_status = str(job.get("status") or "MISSING") result_class = str(result.get("class") or job.get("class") or "MISSING") return { "yeffai_remote_verification_status": remote_status, "remote_verification_status": remote_status, "category_remote_verification_status": remote_status, "yeffai_category_remote_status": remote_status, "remote_execution_status": remote_status, "operational_execution_status": remote_status, "worker_status": job_status, "job_status": job_status, "last_job_status": job_status, "execution_job_status": job_status, "execution_status_remote": remote_status, "category_remote_verify_job_id": str(job.get("job_id") or job.get("id") or "MISSING"), "category_remote_verify_class": result_class, "category_remote_verify_categories_total": int(result.get("categories_total") or 0), "category_remote_verify_categories_missing": int(result.get("categories_missing") or 0), "category_remote_verify_mutation": str(result.get("remote_mutation_performed") or "NO"), } def category_row(phase, category, resolved_map): phase_id = phase.get("phase_id") or category.get("source_phase_id") or category.get("phase_id") or "" category_id = category.get("category_id") or category.get("source_category_id") or "" category_title = category.get("category_title") or category.get("title") or category_id category_contract_path = get_category_contract_path(category) resolved_entry = resolved_map.get(str(category_id)) resolved_state_path = str(resolved_entry["path"]) if resolved_entry else str(STATE_DIR / f"{category_id}.resolved.json") resolved_state_status = get_resolved_status(category_id, resolved_map) progress_percent = get_resolved_progress(category_id, resolved_map, resolved_state_status) item_total = count_item_contracts_for_category(category_id) item_pass_count = item_total if resolved_state_status == "PASS" else 0 category_remote_runtime = category_remote_verify_runtime_fields() next_action = ( category.get("next_contract_action") or category.get("next_required_action") or "NONE" ) if resolved_state_status != "PASS" and next_action == "NONE": next_action = "resolve_missing_or_failed_state" return { "phase_id": phase_id, "category_id": category_id, "category_title": category_title, "category_contract_status": normalize_contract_status(category.get("category_contract_status")), "execution_items_status": normalize_contract_status(category.get("execution_items_status")), "monitoring_items_status": normalize_contract_status(category.get("monitoring_items_status")), "progress_rules_status": normalize_contract_status(category.get("progress_rules_status")), "scan_contract_status": normalize_contract_status(category.get("scan_contract_status")), "refresh_contract_status": normalize_contract_status(category.get("refresh_contract_status")), "resolved_state_status": resolved_state_status, "yeffai_remote_verification_status": category_remote_runtime["yeffai_remote_verification_status"], "remote_verification_status": category_remote_runtime["remote_verification_status"], "category_remote_verification_status": category_remote_runtime["category_remote_verification_status"], "yeffai_category_remote_status": category_remote_runtime["yeffai_category_remote_status"], "remote_execution_status": category_remote_runtime["remote_execution_status"], "operational_execution_status": category_remote_runtime["operational_execution_status"], "worker_status": category_remote_runtime["worker_status"], "job_status": category_remote_runtime["job_status"], "last_job_status": category_remote_runtime["last_job_status"], "execution_job_status": category_remote_runtime["execution_job_status"], "execution_status_remote": category_remote_runtime["execution_status_remote"], "category_remote_verify_job_id": category_remote_runtime["category_remote_verify_job_id"], "category_remote_verify_class": category_remote_runtime["category_remote_verify_class"], "category_remote_verify_categories_total": category_remote_runtime["category_remote_verify_categories_total"], "category_remote_verify_categories_missing": category_remote_runtime["category_remote_verify_categories_missing"], "category_remote_verify_mutation": category_remote_runtime["category_remote_verify_mutation"], "progress_percent": progress_percent, "item_pass_count": item_pass_count, "item_total": item_total, "scan_status": normalize_final_status(category.get("scan_status") or category.get("scan_contract_status")), "refresh_status": normalize_final_status(category.get("refresh_status") or category.get("refresh_contract_status")), "blocking_reasons": get_blocking_reasons(category_id, resolved_map), "source_path": category.get("source_path") or "", "category_contract_path": category_contract_path, "resolved_state_path": resolved_state_path, "next_contract_action": next_action, } def phase_row(phase, categories): category_count = len(categories) active_count = sum(1 for c in categories if c["category_contract_status"] == "ACTIVE") pass_count = sum(1 for c in categories if c["resolved_state_status"] == "PASS") fail_count = sum(1 for c in categories if c["resolved_state_status"] == "FAIL") missing_count = sum(1 for c in categories if c["resolved_state_status"] == "MISSING") if category_count: progress_percent = round(sum(c["progress_percent"] for c in categories) / category_count) else: progress_percent = 0 status = "PASS" if fail_count: status = "FAIL" elif missing_count or category_count == 0: status = "MISSING" return { "phase_id": phase.get("phase_id") or "", "phase_title": phase.get("phase_title") or phase.get("title") or phase.get("phase_id") or "", "phase_order": phase.get("phase_order") or phase.get("order") or 0, "category_count": category_count, "active_count": active_count, "pass_count": pass_count, "fail_count": fail_count, "missing_count": missing_count, "progress_percent": progress_percent, "status": status, "categories": categories, } def build(): master = read_json(MASTER_PATH, required=True) validation_report = read_json(VALIDATION_REPORT_PATH) generic_summary = read_json(GENERIC_SUMMARY_PATH) expansion_plan = read_json(EXPANSION_PLAN_PATH) bulk_report = read_json(BULK_REPORT_PATH) phases_src = master.get("phases", []) if not isinstance(phases_src, list): raise ValueError("master_execution_contract.json does not contain phases[]") resolved_map = build_resolved_state_map() phases_out = [] for phase in phases_src: categories_src = phase.get("categories", []) if not isinstance(categories_src, list): categories_src = [] categories_out = [ category_row(phase, category, resolved_map) for category in categories_src ] phases_out.append(phase_row(phase, categories_out)) all_categories = [category for phase in phases_out for category in phase["categories"]] phase_count = len(phases_out) category_count = len(all_categories) active_category_count = sum(1 for c in all_categories if c["category_contract_status"] == "ACTIVE") resolved_category_count = sum(1 for c in all_categories if c["resolved_state_status"] in {"PASS", "FAIL", "MISSING"}) pass_count = sum(1 for c in all_categories if c["resolved_state_status"] == "PASS") fail_count = sum(1 for c in all_categories if c["resolved_state_status"] == "FAIL") missing_count = sum(1 for c in all_categories if c["resolved_state_status"] == "MISSING") validator_status = get_report_status(validation_report) planner_status = get_planner_status(expansion_plan) expansion_status = get_expansion_status(expansion_plan) candidate_category_count = get_candidate_category_count(expansion_plan) if candidate_category_count == 0: candidate_category_count = get_candidate_category_count(bulk_report) forbidden_path_findings = find_forbidden_path_findings() global_progress = 0 if category_count: global_progress = round(sum(c["progress_percent"] for c in all_categories) / category_count) critical_categories = [ { "phase_id": c["phase_id"], "category_id": c["category_id"], "category_title": c["category_title"], "status": c["resolved_state_status"], "blocking_reasons": c["blocking_reasons"], "next_contract_action": c["next_contract_action"], } for c in all_categories if c["resolved_state_status"] in {"FAIL", "MISSING"} ] blocked_categories = [ { "phase_id": c["phase_id"], "category_id": c["category_id"], "category_title": c["category_title"], "blocking_reasons": c["blocking_reasons"], "next_contract_action": c["next_contract_action"], } for c in all_categories if c["blocking_reasons"] ] # DEVON_EXACT_BOOTSTRAP_NEXT_ACTION_FROM_PLAN_LEDGER_V1 plan_next_action = "" ledger_required_next_action = "" plan_path = STATE_DIR / "execution_server_integration_plan.json" ledger_path = STATE_DIR / "execution_ledger.json" if plan_path.exists(): try: plan_data = json.loads(plan_path.read_text(encoding="utf-8")) if isinstance(plan_data, dict): plan_next_action = str(plan_data.get("next_required_action") or "").strip() except Exception: plan_next_action = "" if ledger_path.exists(): try: ledger_data = json.loads(ledger_path.read_text(encoding="utf-8")) if isinstance(ledger_data, dict): ledger_required_next_action = str(ledger_data.get("required_next_action") or "").strip() except Exception: ledger_required_next_action = "" if plan_next_action: next_action = plan_next_action elif ledger_required_next_action: next_action = ledger_required_next_action elif fail_count or missing_count: next_action = "RESOLVE_FAIL_OR_MISSING_CATEGORIES" elif validator_status != "PASS": next_action = "FIX_MASTER_EXECUTION_VALIDATION" elif planner_status != "PASS" or expansion_status != "COMPLETE": next_action = "FIX_CONTRACT_EXPANSION_PLANNER" else: next_action = "MISSING_CANONICAL_NEXT_ACTION" output = { "meta": { "generated_at": utc_now(), "source_root": str(DEVON_ROOT), "canon_execution_root": str(EXEC_ROOT), "version": "panel_bootstrap_index.v1", "status": "PASS" if not forbidden_path_findings else "FAIL", "source_hashes": { "master_execution_contract": sha256_file(MASTER_PATH), "validation_report": sha256_file(VALIDATION_REPORT_PATH), "generic_summary": sha256_file(GENERIC_SUMMARY_PATH), "expansion_plan": sha256_file(EXPANSION_PLAN_PATH), "bulk_report": sha256_file(BULK_REPORT_PATH), }, }, "global_summary": { "phase_count": phase_count, "category_count": category_count, "active_category_count": active_category_count, "candidate_category_count": candidate_category_count, "resolved_category_count": resolved_category_count, "pass_count": pass_count, "fail_count": fail_count, "missing_count": missing_count, "json_file_count": count_json_files_excluding_output(), "validator_status": validator_status, "planner_status": planner_status, "expansion_status": expansion_status, "forbidden_path_findings": len(forbidden_path_findings), "active_legacy_panel_dir_status": "FORBIDDEN_NOT_USED" if not forbidden_path_findings else "FORBIDDEN_PATH_FOUND", }, "phases": phases_out, "navigation": { "phases": [ { "phase_id": p["phase_id"], "phase_title": p["phase_title"], "phase_order": p["phase_order"], "status": p["status"], "progress_percent": p["progress_percent"], "categories": [ { "category_id": c["category_id"], "category_title": c["category_title"], "status": c["resolved_state_status"], "progress_percent": c["progress_percent"], } for c in p["categories"] ], } for p in phases_out ] }, "dashboard": { "global_progress": global_progress, "global_cards": { "phases": phase_count, "categories": category_count, "active_categories": active_category_count, "pass": pass_count, "fail": fail_count, "missing": missing_count, "json_files": count_json_files_excluding_output(), }, "phase_donuts": [ { "phase_id": p["phase_id"], "phase_title": p["phase_title"], "status": p["status"], "progress_percent": p["progress_percent"], "pass_count": p["pass_count"], "fail_count": p["fail_count"], "missing_count": p["missing_count"], "category_count": p["category_count"], } for p in phases_out ], "critical_categories": critical_categories, "blocked_categories": blocked_categories, "next_action": next_action, "validator": { "status": validator_status, "path": str(VALIDATION_REPORT_PATH), }, "planner": { "status": planner_status, "expansion_status": expansion_status, "path": str(EXPANSION_PLAN_PATH), }, }, "contract_sources": { "master_execution_contract": str(MASTER_PATH), "validation_report": str(VALIDATION_REPORT_PATH), "generic_summary": str(GENERIC_SUMMARY_PATH), "expansion_plan": str(EXPANSION_PLAN_PATH), "bulk_report": str(BULK_REPORT_PATH), }, "rules": { "frontend_must_not_infer_status": True, "frontend_must_not_calculate_progress_from_raw_files": True, "frontend_reads_panel_bootstrap_index_first": True, "canon_decides_panel_renders": True, "legacy_panel_path_forbidden": True, "panel_category_names_allowed": True, "allowed_statuses": ["PASS", "FAIL", "MISSING"], "legacy_panel_path_policy": "absolute_legacy_panel_path_forbidden", }, } rendered = json.dumps(output, ensure_ascii=False, indent=2, sort_keys=False) if FORBIDDEN_ABSOLUTE_PANEL_PATH in rendered: raise ValueError("Generated panel_bootstrap_index contains forbidden absolute legacy panel path") tmp = OUTPUT_PATH.with_suffix(".json.tmp") tmp.write_text(rendered + "\n", encoding="utf-8") tmp.replace(OUTPUT_PATH) return output def validate_output(output): summary = output["global_summary"] failures = [] expected = { "phase_count": 14, "category_count": 190, "active_category_count": 190, "resolved_category_count": 190, "pass_count": 152, "fail_count": 0, "missing_count": 38, "candidate_category_count": 0, "expansion_status": "COMPLETE", "forbidden_path_findings": 0, } for key, expected_value in expected.items(): actual = summary.get(key) if actual != expected_value: failures.append(f"{key}: expected={expected_value} actual={actual}") if summary.get("validator_status") != "PASS": failures.append(f"validator_status: expected=PASS actual={summary.get('validator_status')}") if summary.get("planner_status") != "PASS": failures.append(f"planner_status: expected=PASS actual={summary.get('planner_status')}") raw = OUTPUT_PATH.read_text(encoding="utf-8", errors="ignore") if FORBIDDEN_ABSOLUTE_PANEL_PATH in raw: failures.append("raw forbidden absolute /panel path found in panel_bootstrap_index.json") # resolved_completion_allowed_pass_missing_pairs # Valid states for Phase 11-14 after contract expansion: # - contract-ready unresolved: 152 PASS / 38 MISSING # - resolved-complete: 190 PASS / 0 MISSING actual_pass = output.get("global_summary", {}).get("pass_count") actual_missing = output.get("global_summary", {}).get("missing_count") allowed_pass_missing_pairs = [(152, 38), (190, 0)] if (actual_pass, actual_missing) in allowed_pass_missing_pairs: failures = [ failure for failure in failures if not ( isinstance(failure, str) and ( failure.startswith("pass_count:") or failure.startswith("missing_count:") ) ) ] else: failures.append( "pass_missing_pair: expected_one_of=" + str(allowed_pass_missing_pairs) + " actual=" + str((actual_pass, actual_missing)) ) if failures: return False, failures return True, [] if __name__ == "__main__": built = build() ok, errors = validate_output(built) print("PANEL_BOOTSTRAP_INDEX_PATH=" + str(OUTPUT_PATH)) print("PANEL_BOOTSTRAP_INDEX_JSON=PASS") print("PHASE_COUNT=" + str(built["global_summary"]["phase_count"])) print("CATEGORY_COUNT=" + str(built["global_summary"]["category_count"])) print("ACTIVE_CATEGORY_COUNT=" + str(built["global_summary"]["active_category_count"])) print("RESOLVED_CATEGORY_COUNT=" + str(built["global_summary"]["resolved_category_count"])) print("PASS_COUNT=" + str(built["global_summary"]["pass_count"])) print("FAIL_COUNT=" + str(built["global_summary"]["fail_count"])) print("MISSING_COUNT=" + str(built["global_summary"]["missing_count"])) print("CANDIDATE_CATEGORY_COUNT=" + str(built["global_summary"]["candidate_category_count"])) print("JSON_FILE_COUNT=" + str(built["global_summary"]["json_file_count"])) print("VALIDATOR_STATUS=" + str(built["global_summary"]["validator_status"])) print("PLANNER_STATUS=" + str(built["global_summary"]["planner_status"])) print("EXPANSION_STATUS=" + str(built["global_summary"]["expansion_status"])) print("FORBIDDEN_PATH_FINDINGS=" + str(built["global_summary"]["forbidden_path_findings"])) if ok: print("FINAL_VALIDATION=PASS") else: print("FINAL_VALIDATION=FAIL") for error in errors: print("VALIDATION_ERROR=" + error)