#!/usr/bin/env python3 import json import subprocess from pathlib import Path from datetime import datetime, timezone ROOT = Path("/home/yeff/public_html/devon") CANON_EXEC = ROOT / "canon" / "execution" MASTER = CANON_EXEC / "master_execution_contract.json" STATE_DIR = CANON_EXEC / "state" ITEMS_DIR = CANON_EXEC / "items" SCANS_DIR = CANON_EXEC / "scans" REFRESH_DIR = CANON_EXEC / "refresh" PROGRESS_DIR = CANON_EXEC / "progress" FORBIDDEN_ABSOLUTE_PATHS = [ str(ROOT / "panel"), str(ROOT / "panel" / "data"), ] def utc_now(): return datetime.now(timezone.utc).isoformat() def read_json(path): return json.loads(path.read_text(encoding="utf-8")) def write_json(path, data): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def run_command(command, timeout=30): if not command or not command.strip(): return { "status": "MISSING", "exit_code": None, "stdout": "", "stderr": "missing command" } try: result = subprocess.run( command, shell=True, cwd=str(ROOT), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, ) stdout = (result.stdout or "").strip() stderr = (result.stderr or "").strip() if result.returncode != 0: status = "FAIL" elif "FAIL" in stdout: status = "FAIL" else: status = "PASS" return { "status": status, "exit_code": result.returncode, "stdout": stdout[-4000:], "stderr": stderr[-4000:] } except subprocess.TimeoutExpired as exc: return { "status": "FAIL", "exit_code": None, "stdout": (exc.stdout or "")[-4000:] if isinstance(exc.stdout, str) else "", "stderr": "command timeout" } except Exception as exc: return { "status": "FAIL", "exit_code": None, "stdout": "", "stderr": str(exc) } def collect_active_categories(): master = read_json(MASTER) rows = [] for phase in master.get("phases", []): for category in phase.get("categories", []): if category.get("category_contract_status") != "ACTIVE": continue category_id = category.get("category_id", "") contract_path = Path(category.get("category_contract_path", "")) if not category_id or not contract_path.exists(): continue rows.append({ "phase_id": phase.get("phase_id", ""), "category_id": category_id, "category_title": category.get("category_title", ""), "category_contract_path": contract_path }) return rows def collect_category_files(category_id, category_contract_path): item_files = sorted(ITEMS_DIR.glob(f"{category_id}_*.contract.json")) scan_file = SCANS_DIR / f"{category_id}_category_scan.contract.json" refresh_file = REFRESH_DIR / f"{category_id}_refresh.contract.json" progress_file = PROGRESS_DIR / f"{category_id}_category_readiness.contract.json" required = [category_contract_path, scan_file, refresh_file, progress_file] + item_files return { "item_files": item_files, "scan_file": scan_file, "refresh_file": refresh_file, "progress_file": progress_file, "required_files": required } def resolve_item(path): data = read_json(path) item_id = data.get("item_id", path.stem) evidence_command = data.get("evidence", {}).get("evidence_command", "") validation_command = data.get("validation", {}).get("validation_command", "") timeout = data.get("validation", {}).get("validation_timeout_seconds", 30) evidence_result = run_command(evidence_command, timeout) validation_result = run_command(validation_command, timeout) checkpoints = data.get("progress", {}).get("checkpoints", []) checkpoint_results = [] progress_percent = 0 for checkpoint in checkpoints: command = checkpoint.get("evidence_command", "") result = run_command(command, 30) status = result["status"] checkpoint_results.append({ "percent": checkpoint.get("percent", 0), "condition": checkpoint.get("condition", ""), "status": status, "stdout": result["stdout"], "stderr": result["stderr"] }) if status == "PASS": progress_percent = max(progress_percent, int(checkpoint.get("percent", 0))) final_status = "PASS" if evidence_result["status"] == "PASS" and validation_result["status"] == "PASS" else "FAIL" return { "item_id": item_id, "item_contract_path": str(path), "evidence_status": evidence_result["status"], "validation_status": validation_result["status"], "final_status": final_status, "progress_percent": progress_percent, "evidence_stdout": evidence_result["stdout"], "evidence_stderr": evidence_result["stderr"], "validation_stdout": validation_result["stdout"], "validation_stderr": validation_result["stderr"], "checkpoint_results": checkpoint_results, "missing_to_100": [] if final_status == "PASS" and progress_percent == 100 else data.get("progress", {}).get("missing_to_100", []) } def resolve_progress(path): if not path.exists(): return { "progress_contract_path": str(path), "current_percent": 0, "final_status": "MISSING", "checkpoint_results": [], "missing_to_100": [ { "missing_id": "progress_contract_missing", "description": "Progress contract file is missing.", "required_evidence": str(path), "blocking": True, "next_action": "create_progress_contract" } ] } data = read_json(path) checkpoints = data.get("checkpoints", []) current_percent = 0 results = [] for checkpoint in checkpoints: command = checkpoint.get("evidence_command", "") result = run_command(command, 30) status = result["status"] results.append({ "percent": checkpoint.get("percent", 0), "condition": checkpoint.get("condition", ""), "status": status, "stdout": result["stdout"], "stderr": result["stderr"] }) if status == "PASS": current_percent = max(current_percent, int(checkpoint.get("percent", 0))) final_status = "PASS" if current_percent == 100 else ("MISSING" if current_percent == 0 else "FAIL") return { "progress_contract_path": str(path), "current_percent": current_percent, "final_status": final_status, "checkpoint_results": results, "missing_to_100": [] if current_percent == 100 else data.get("missing_to_100", []) } def resolve_category(row): category_id = row["category_id"] files = collect_category_files(category_id, row["category_contract_path"]) missing_files = [str(p) for p in files["required_files"] if not p.exists()] json_errors = [] serialized = "" for path in files["required_files"]: if not path.exists(): continue try: text = path.read_text(encoding="utf-8") serialized += text json.loads(text) except Exception as exc: json_errors.append({ "path": str(path), "error": str(exc) }) forbidden_found = [x for x in FORBIDDEN_ABSOLUTE_PATHS if x in serialized] item_results = [] for path in files["item_files"]: try: item_results.append(resolve_item(path)) except Exception as exc: item_results.append({ "item_id": path.stem, "item_contract_path": str(path), "final_status": "FAIL", "progress_percent": 0, "error": str(exc) }) progress_result = resolve_progress(files["progress_file"]) scan_result = { "status": "MISSING", "stdout": "", "stderr": "", "exit_code": None } if files["scan_file"].exists(): scan_contract = read_json(files["scan_file"]) scan_result = run_command(scan_contract.get("scan_command", ""), 30) refresh_checks = [] if files["refresh_file"].exists(): refresh_contract = read_json(files["refresh_file"]) for command in refresh_contract.get("checks_to_run", []): result = run_command(command, 30) refresh_checks.append({ "command": command, "status": result["status"], "stdout": result["stdout"], "stderr": result["stderr"] }) item_total = len(item_results) item_pass_count = sum(1 for item in item_results if item.get("final_status") == "PASS") blocking_reasons = [] if missing_files: blocking_reasons.append("required_contract_file_missing") if json_errors: blocking_reasons.append("json_parse_error") if forbidden_found: blocking_reasons.append("forbidden_absolute_path_found") if item_total == 0 or item_pass_count < item_total: blocking_reasons.append("one_or_more_items_not_pass") if progress_result.get("current_percent", 0) < 100: blocking_reasons.append("progress_not_100") if scan_result["status"] != "PASS": blocking_reasons.append("category_scan_failed") refresh_status = "PASS" if refresh_checks and all(x["status"] == "PASS" for x in refresh_checks) else "FAIL" if refresh_status != "PASS": blocking_reasons.append("refresh_check_failed") final_status = "PASS" if not blocking_reasons else "FAIL" return { "state_id": f"resolved_state.{category_id}", "version": "v1", "resolved_at": utc_now(), "phase_id": row["phase_id"], "category_id": category_id, "category_title": row["category_title"], "category_contract_path": str(row["category_contract_path"]), "state_source": "generic_category_state_resolver_v1", "final_status": final_status, "progress_percent": progress_result.get("current_percent", 0), "item_pass_count": item_pass_count, "item_total": item_total, "scan_status": scan_result["status"], "refresh_status": refresh_status, "missing_files": missing_files, "json_errors": json_errors, "forbidden_absolute_paths_found": forbidden_found, "blocking_reasons": blocking_reasons, "item_results": item_results, "scan_result": scan_result, "refresh_checks": refresh_checks, "progress_result": progress_result, "next_required_action": "create_next_category_execution_contract" if final_status == "PASS" else "fix_failed_contract_or_evidence" } def main(): STATE_DIR.mkdir(parents=True, exist_ok=True) rows = collect_active_categories() results = [] for row in rows: state = resolve_category(row) write_json(STATE_DIR / f"{row['category_id']}.resolved.json", state) results.append(state) final_pass = sum(1 for row in results if row.get("final_status") == "PASS") final_fail = sum(1 for row in results if row.get("final_status") == "FAIL") summary = { "summary_id": "generic_category_state_resolution_summary", "version": "v1", "resolved_at": utc_now(), "resolved_category_count": len(results), "pass_count": final_pass, "fail_count": final_fail, "results": [ { "category_id": row.get("category_id"), "final_status": row.get("final_status"), "progress_percent": row.get("progress_percent"), "item_pass_count": row.get("item_pass_count"), "item_total": row.get("item_total"), "blocking_reasons": row.get("blocking_reasons", []) } for row in results ], "final_status": "PASS" if final_fail == 0 and len(results) > 0 else "FAIL", "next_required_action": "run_contract_validator_and_planner" } write_json(STATE_DIR / "generic_category_state_resolution_summary.json", summary) print("GENERIC_CATEGORY_STATE_RESOLVER_STATUS=PASS") print("RESOLVED_CATEGORY_COUNT=", len(results)) print("PASS_COUNT=", final_pass) print("FAIL_COUNT=", final_fail) print("SUMMARY=", STATE_DIR / "generic_category_state_resolution_summary.json") for row in results: print( "CATEGORY=", row.get("category_id"), "| FINAL=", row.get("final_status"), "| PROGRESS=", row.get("progress_percent"), "| ITEMS=", f"{row.get('item_pass_count')}/{row.get('item_total')}", "| BLOCKING=", ",".join(row.get("blocking_reasons", [])) ) if __name__ == "__main__": main()